/* -------------------------------------------------------------------------
 *
 * arrayfuncs.c
 *	  Support functions for arrays.
 *
 * Portions Copyright (c) 1996-2012, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  src/backend/utils/adt/arrayfuncs.c
 *
 * -------------------------------------------------------------------------
 */
#include "postgres.h"
#include "knl/knl_variable.h"

#include <ctype.h>

#include "catalog/pg_proc.h"
#include "common/int.h"
#include "funcapi.h"
#include "libpq/pqformat.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/datum.h"
#include "utils/extended_statistics.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/typcache.h"
#include "nodes/nodes.h"
#include "nodes/nodeFuncs.h"

/*
 * Local definitions
 */
#define ASSGN "="

typedef enum {
    ARRAY_NO_LEVEL,
    ARRAY_LEVEL_STARTED,
    ARRAY_ELEM_STARTED,
    ARRAY_ELEM_COMPLETED,
    ARRAY_QUOTED_ELEM_STARTED,
    ARRAY_QUOTED_ELEM_COMPLETED,
    ARRAY_ELEM_DELIMITED,
    ARRAY_LEVEL_COMPLETED,
    ARRAY_LEVEL_DELIMITED
} ArrayParseState;

/* Working state for array_iterate() */
typedef struct ArrayIteratorData {
    /* basic info about the array, set up during array_create_iterator() */
    ArrayType* arr;    /* array we're iterating through */
    bits8* nullbitmap; /* its null bitmap, if any */
    int nitems;        /* total number of elements in array */
    int16 typlen;      /* element type's length */
    bool typbyval;     /* element type's byval property */
    char typalign;     /* element type's align property */

    /* information about the requested slice size */
    int slice_ndim;      /* slice dimension, or 0 if not slicing */
    int slice_len;       /* number of elements per slice */
    int* slice_dims;     /* slice dims array */
    int* slice_lbound;   /* slice lbound array */
    Datum* slice_values; /* workspace of length slice_len */
    bool* slice_nulls;   /* workspace of length slice_len */

    /* current position information, updated on each iteration */
    char* data_ptr;   /* our current position in the array */
    int current_item; /* the item # we're at in the array */
} ArrayIteratorData;

typedef struct ArrayItem {
    Datum elt;
    bool isnull;
} ArrayItem;

typedef struct CmpFuncArgs {
    FunctionCallInfoData locfcinfo;
    FmgrInfo func;
    Oid collation;
} CmpFuncArgs;

static bool array_isspace(char ch);
static int ArrayCount(const char* str, int* dim, char typdelim);
static void ReadArrayStr(char* arrayStr, const char* origStr, int nitems, int ndim, const int* dim, FmgrInfo* inputproc,
    Oid typioparam, int32 typmod, char typdelim, int typlen, bool typbyval, char typalign, Datum* values, bool* nulls,
    bool* hasnulls, int32* nbytes);
static void ReadArrayBinary(StringInfo buf, int nitems, FmgrInfo* receiveproc, Oid typioparam, int32 typmod, int typlen,
    bool typbyval, char typalign, Datum* values, bool* nulls, bool* hasnulls, int32* nbytes);
static bool array_get_isnull(const bits8* nullbitmap, int offset);
static void array_set_isnull(bits8* nullbitmap, int offset, bool isNull);
static Datum ArrayCast(const char* value, bool byval, int len);
static char* array_seek(char* ptr, int offset, bits8* nullbitmap, int nitems, int typlen, bool typbyval, char typalign);
static int array_nelems_size(
    char* ptr, int offset, bits8* nullbitmap, int nitems, int typlen, bool typbyval, char typalign);
static int array_copy(
    char* destptr, int nitems, char* srcptr, int offset, bits8* nullbitmap, int typlen, bool typbyval, char typalign);
static int array_slice_size(char* arraydataptr, bits8* arraynullsptr, int ndim, int* dim, int* lb, const int* st,
    int* endp, int typlen, bool typbyval, char typalign);
static void array_extract_slice(ArrayType* newarray, int ndim, int* dim, int* lb, char* arraydataptr,
    bits8* arraynullsptr, const int* st, int* endp, int typlen, bool typbyval, char typalign);
static void array_insert_slice(ArrayType* destArray, ArrayType* origArray, ArrayType* srcArray, int ndim, int* dim,
    int* lb, const int* st, int* endp, int typlen, bool typbyval, char typalign);
static int array_cmp(FunctionCallInfo fcinfo);
static ArrayType* create_array_envelope(int ndims, int* dimv, const int* lbv, int nbytes, Oid elmtype, int dataoffset);
static ArrayType* array_fill_internal(
    ArrayType* dims, ArrayType* lbs, Datum value, bool isnull, Oid elmtype, FunctionCallInfo fcinfo);
static ArrayType* array_deleteidx_internal(ArrayType *v, int delIndex);
static ArrayType* array_deleteidx_internal_db_a(ArrayType *v, int delIndex1, int delIndex2, bool multi_args);
static void checkEnv();

/*
 * complex_array_in :
 *		  converts an array-in-array from the external format in "string" to
 *		  its internal format.
 *        this function is used to restore multi-column MCV when adding CN nodes
 *
 * return value :
 *		  the internal representation of the input array-in-array
 */
Datum complex_array_in(PG_FUNCTION_ARGS)
{
    char* string = PG_GETARG_CSTRING(0);                    /* external form */
    Oid relid = PG_GETARG_OID(1);                           /* table oid */
    int2vector* stakey = (int2vector*)PG_GETARG_POINTER(2); /* statistic key */
    ArrayType* retval = NULL;

    PG_TRY();
    {
        Oid atttypid = CSTRINGOID;
        const int atttypmod = -1;
        Oid* atttypid_array = NULL;
        int* atttypmod_array = NULL;
        uint32 num_column = 1;
        es_get_columns_typid_typmod(relid, stakey, &atttypid_array, &atttypmod_array, &num_column);

        Datum array_datum = OidFunctionCall3Coll(
            ANYARRAYINFUNCOID, InvalidOid, CStringGetDatum(string), UInt32GetDatum(atttypid), Int32GetDatum(atttypmod));

        array_datum =
            es_mcv_slot_cstring_array_to_array_array(array_datum, num_column, atttypid_array, atttypmod_array);
        pfree(atttypid_array);
        pfree(atttypmod_array);
        atttypid_array = NULL;
        atttypmod_array = NULL;

        retval = (ArrayType*)DatumGetPointer(array_datum);
    }
    PG_CATCH();
    {
        retval = NULL;
        ereport(ERROR,
            (errcode(ERRCODE_SUCCESSFUL_COMPLETION),
                errmsg("Parameters of complex_array_in are invalid."),
                handle_in_client(true)));
        PG_RE_THROW();
    }
    PG_END_TRY();

    PG_RETURN_ARRAYTYPE_P(retval);
}

/*
 * array_in :
 *		  converts an array from the external format in "string" to
 *		  its internal format.
 *
 * return value :
 *		  the internal representation of the input array
 */
Datum array_in(PG_FUNCTION_ARGS)
{
    char* string = PG_GETARG_CSTRING(0); /* external form */
    Oid element_type = PG_GETARG_OID(1); /* type of an array
                                          * element */
    int32 typmod = PG_GETARG_INT32(2);   /* typmod for array elements */
    int typlen;
    bool typbyval = false;
    char typalign;
    char typdelim;
    Oid typioparam;
    char *string_save = NULL, *p = NULL;
    int i, nitems;
    Datum* dataPtr = NULL;
    bool* nullsPtr = NULL;
    bool hasnulls = false;
    int32 nbytes;
    int32 dataoffset;
    ArrayType* retval = NULL;
    int ndim, dim[MAXDIM], lBound[MAXDIM];
    ArrayMetaState* my_extra = NULL;

    /*
     * We arrange to look up info about element type, including its input
     * conversion proc, only once per series of calls, assuming the element
     * type doesn't change underneath us.
     */
    my_extra = (ArrayMetaState*)fcinfo->flinfo->fn_extra;
    if (my_extra == NULL) {
        fcinfo->flinfo->fn_extra = MemoryContextAlloc(fcinfo->flinfo->fn_mcxt, sizeof(ArrayMetaState));
        my_extra = (ArrayMetaState*)fcinfo->flinfo->fn_extra;
        my_extra->element_type = ~element_type;
    }

    if (my_extra->element_type != element_type) {
        /*
         * Get info about element type, including its input conversion proc
         */
        get_type_io_data(element_type,
            IOFunc_input,
            &my_extra->typlen,
            &my_extra->typbyval,
            &my_extra->typalign,
            &my_extra->typdelim,
            &my_extra->typioparam,
            &my_extra->typiofunc);
        fmgr_info_cxt(my_extra->typiofunc, &my_extra->proc, fcinfo->flinfo->fn_mcxt);
        my_extra->element_type = element_type;
    }
    typlen = my_extra->typlen;
    typbyval = my_extra->typbyval;
    typalign = my_extra->typalign;
    typdelim = my_extra->typdelim;
    typioparam = my_extra->typioparam;

    /* Make a modifiable copy of the input */
    string_save = pstrdup(string);

    /*
     * If the input string starts with dimension info, read and use that.
     * Otherwise, we require the input to be in curly-brace style, and we
     * prescan the input to determine dimensions.
     *
     * Dimension info takes the form of one or more [n] or [m:n] items. The
     * outer loop iterates once per dimension item.
     */
    p = string_save;
    ndim = 0;
    for (;;) {
        char* q = NULL;
        int ub;

        /*
         * Note: we currently allow whitespace between, but not within,
         * dimension items.
         */
        while (array_isspace(*p))
            p++;
        if (*p != '[')
            break; /* no more dimension items */
        p++;
        if (ndim >= MAXDIM)
            ereport(ERROR,
                (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                    errmsg("number of array dimensions (%d) exceeds the maximum allowed (%d)", ndim + 1, MAXDIM)));

        for (q = p; isdigit((unsigned char)*q) || (*q == '-') || (*q == '+'); q++) {
            ;
        }
        if (q == p) /* no digits? */
            ereport(ERROR, (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION), errmsg("missing dimension value")));

        if (*q == ':') {
            /* [m:n] format */
            *q = '\0';
            lBound[ndim] = atoi(p);
            p = q + 1;
            for (q = p; isdigit((unsigned char)*q) || (*q == '-') || (*q == '+'); q++) {
                ;
            }
            if (q == p) /* no digits? */
                ereport(ERROR, (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION), errmsg("missing dimension value")));
        } else {
            /* [n] format */
            lBound[ndim] = 1;
        }
        if (*q != ']')
            ereport(ERROR, (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION), errmsg("missing \"]\" in array dimensions")));

        *q = '\0';
        ub = atoi(p);
        p = q + 1;
        if (ub < lBound[ndim])
            ereport(
                ERROR, (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR), errmsg("upper bound cannot be less than lower bound")));

        dim[ndim] = ub - lBound[ndim] + 1;
        ndim++;
    }

    if (ndim == 0) {
        /* No array dimensions, so intuit dimensions from brace structure */
        if (*p != '{')
            ereport(ERROR,
                (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
                    errmsg("array value must start with \"{\" or dimension information")));
        ndim = ArrayCount(p, dim, typdelim);
        for (i = 0; i < ndim; i++)
            lBound[i] = 1;
    } else {
        int ndim_braces, dim_braces[MAXDIM];

        /* If array dimensions are given, expect '=' operator */
        if (strncmp(p, ASSGN, strlen(ASSGN)) != 0)
            ereport(ERROR, (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION), errmsg("missing assignment operator")));
        p += strlen(ASSGN);
        while (array_isspace(*p)) {
            p++;
        }

        /*
         * intuit dimensions from brace structure -- it better match what we
         * were given
         */
        if (*p != '{')
            ereport(ERROR,
                (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
                    errmsg("array value must start with \"{\" or dimension information")));
        ndim_braces = ArrayCount(p, dim_braces, typdelim);
        if (ndim_braces != ndim)
            ereport(ERROR,
                (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
                    errmsg("array dimensions incompatible with array literal")));
        for (i = 0; i < ndim; ++i) {
            if (dim[i] != dim_braces[i])
                ereport(ERROR,
                    (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
                        errmsg("array dimensions incompatible with array literal")));
        }
    }

#ifdef ARRAYDEBUG
    printf("array_in- ndim %d (", ndim);
    for (i = 0; i < ndim; i++) {
        printf(" %d", dim[i]);
    };
    printf(") for %s\n", string);
#endif

    /* This checks for overflow of the array dimensions */
    nitems = ArrayGetNItems(ndim, dim);
    ArrayCheckBounds(ndim, dim, lBound);

    /* Empty array? */
    if (nitems == 0)
        PG_RETURN_ARRAYTYPE_P(construct_empty_array(element_type));

    dataPtr = (Datum*)palloc(nitems * sizeof(Datum));
    nullsPtr = (bool*)palloc(nitems * sizeof(bool));
    ReadArrayStr(p,
        string,
        nitems,
        ndim,
        dim,
        &my_extra->proc,
        typioparam,
        typmod,
        typdelim,
        typlen,
        typbyval,
        typalign,
        dataPtr,
        nullsPtr,
        &hasnulls,
        &nbytes);
    if (hasnulls) {
        dataoffset = ARR_OVERHEAD_WITHNULLS(ndim, nitems);
        nbytes += dataoffset;
    } else {
        dataoffset = 0; /* marker for no null bitmap */
        nbytes += ARR_OVERHEAD_NONULLS(ndim);
    }
    retval = (ArrayType*)palloc0(nbytes);
    SET_VARSIZE(retval, nbytes);
    retval->ndim = ndim;
    retval->dataoffset = dataoffset;

    /*
     * This comes from the array's pg_type.typelem (which points to the base
     * data type's pg_type.oid) and stores system oids in user tables. This
     * oid must be preserved by binary upgrades.
     */
    retval->elemtype = element_type;
    errno_t errorno = EOK;
    errorno = memcpy_s(ARR_DIMS(retval), ndim * sizeof(int), dim, ndim * sizeof(int));
    securec_check(errorno, "\0", "\0");
    errorno = memcpy_s(ARR_LBOUND(retval), ndim * sizeof(int), lBound, ndim * sizeof(int));
    securec_check(errorno, "\0", "\0");

    CopyArrayEls(retval, dataPtr, nullsPtr, nitems, typlen, typbyval, typalign, true);

    pfree(dataPtr);
    pfree(nullsPtr);
    pfree(string_save);

    PG_RETURN_ARRAYTYPE_P(retval);
}

/*
 * array_isspace() --- a non-locale-dependent isspace()
 *
 * We used to use isspace() for parsing array values, but that has
 * undesirable results: an array value might be silently interpreted
 * differently depending on the locale setting.  Now we just hard-wire
 * the traditional ASCII definition of isspace().
 */
static bool array_isspace(char ch)
{
    if (ch == ' ' || ch == '\t' || ch == '\n' || ch == '\r' || ch == '\v' || ch == '\f')
        return true;
    return false;
}

/*
 * ArrayCount
 *	 Determines the dimensions for an array string.
 *
 * Returns number of dimensions as function result.  The axis lengths are
 * returned in dim[], which must be of size MAXDIM.
 */
static int ArrayCount(const char* str, int* dim, char typdelim)
{
    int nest_level = 0, i;
    int ndim = 1, temp[MAXDIM], nelems[MAXDIM], nelems_last[MAXDIM];
    bool in_quotes = false;
    bool eoArray = false;
    bool empty_array = true;
    const char* ptr = NULL;
    ArrayParseState parse_state = ARRAY_NO_LEVEL;

    for (i = 0; i < MAXDIM; ++i) {
        temp[i] = dim[i] = 0;
        nelems_last[i] = nelems[i] = 1;
    }

    ptr = str;
    while (!eoArray) {
        bool itemdone = false;

        while (!itemdone) {
            if (parse_state == ARRAY_ELEM_STARTED || parse_state == ARRAY_QUOTED_ELEM_STARTED)
                empty_array = false;

            switch (*ptr) {
                case '\0':
                    /* Signal a premature end of the string */
                    ereport(ERROR,
                        (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION), errmsg("malformed array literal: \"%s\"", str)));
                    break;
                case '\\':

                    /*
                     * An escape must be after a level start, after an element
                     * start, or after an element delimiter. In any case we
                     * now must be past an element start.
                     */
                    if (parse_state != ARRAY_LEVEL_STARTED && parse_state != ARRAY_ELEM_STARTED &&
                        parse_state != ARRAY_QUOTED_ELEM_STARTED && parse_state != ARRAY_ELEM_DELIMITED)
                        ereport(ERROR,
                            (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
                                errmsg("malformed array literal: \"%s\"", str)));
                    if (parse_state != ARRAY_QUOTED_ELEM_STARTED)
                        parse_state = ARRAY_ELEM_STARTED;
                    /* skip the escaped character */
                    if (*(ptr + 1))
                        ptr++;
                    else
                        ereport(ERROR,
                            (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
                                errmsg("malformed array literal: \"%s\"", str)));
                    break;
                case '\"':

                    /*
                     * A quote must be after a level start, after a quoted
                     * element start, or after an element delimiter. In any
                     * case we now must be past an element start.
                     */
                    if (parse_state != ARRAY_LEVEL_STARTED && parse_state != ARRAY_QUOTED_ELEM_STARTED &&
                        parse_state != ARRAY_ELEM_DELIMITED)
                        ereport(ERROR,
                            (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
                                errmsg("malformed array literal: \"%s\"", str)));
                    in_quotes = !in_quotes;
                    if (in_quotes)
                        parse_state = ARRAY_QUOTED_ELEM_STARTED;
                    else
                        parse_state = ARRAY_QUOTED_ELEM_COMPLETED;
                    break;
                case '{':
                    if (!in_quotes) {
                        /*
                         * A left brace can occur if no nesting has occurred
                         * yet, after a level start, or after a level
                         * delimiter.
                         */
                        if (parse_state != ARRAY_NO_LEVEL && parse_state != ARRAY_LEVEL_STARTED &&
                            parse_state != ARRAY_LEVEL_DELIMITED)
                            ereport(ERROR,
                                (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
                                    errmsg("malformed array literal: \"%s\"", str)));
                        parse_state = ARRAY_LEVEL_STARTED;
                        if (nest_level >= MAXDIM)
                            ereport(ERROR,
                                (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                                    errmsg("number of array dimensions (%d) exceeds the maximum allowed (%d)",
                                        nest_level + 1,
                                        MAXDIM)));
                        temp[nest_level] = 0;
                        nest_level++;
                        if (ndim < nest_level)
                            ndim = nest_level;
                    }
                    break;
                case '}':
                    if (!in_quotes) {
                        /*
                         * A right brace can occur after an element start, an
                         * element completion, a quoted element completion, or
                         * a level completion.
                         */
                        if (parse_state != ARRAY_ELEM_STARTED && parse_state != ARRAY_ELEM_COMPLETED &&
                            parse_state != ARRAY_QUOTED_ELEM_COMPLETED && parse_state != ARRAY_LEVEL_COMPLETED &&
                            !(nest_level == 1 && parse_state == ARRAY_LEVEL_STARTED))
                            ereport(ERROR,
                                (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
                                    errmsg("malformed array literal: \"%s\"", str)));
                        parse_state = ARRAY_LEVEL_COMPLETED;
                        if (nest_level == 0)
                            ereport(ERROR,
                                (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
                                    errmsg("malformed array literal: \"%s\"", str)));
                        nest_level--;

                        if ((nelems_last[nest_level] != 1) && (nelems[nest_level] != nelems_last[nest_level]))
                            ereport(ERROR,
                                (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
                                    errmsg("multidimensional arrays must have "
                                           "array expressions with matching "
                                           "dimensions")));
                        nelems_last[nest_level] = nelems[nest_level];
                        nelems[nest_level] = 1;
                        if (nest_level == 0)
                            eoArray = itemdone = true;
                        else {
                            /*
                             * We don't set itemdone here; see comments in
                             * ReadArrayStr
                             */
                            temp[nest_level - 1]++;
                        }
                    }
                    break;
                default:
                    if (!in_quotes) {
                        if (*ptr == typdelim) {
                            /*
                             * Delimiters can occur after an element start, an
                             * element completion, a quoted element
                             * completion, or a level completion.
                             */
                            if (parse_state != ARRAY_ELEM_STARTED && parse_state != ARRAY_ELEM_COMPLETED &&
                                parse_state != ARRAY_QUOTED_ELEM_COMPLETED && parse_state != ARRAY_LEVEL_COMPLETED)
                                ereport(ERROR,
                                    (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
                                        errmsg("malformed array literal: \"%s\"", str)));
                            if (parse_state == ARRAY_LEVEL_COMPLETED)
                                parse_state = ARRAY_LEVEL_DELIMITED;
                            else
                                parse_state = ARRAY_ELEM_DELIMITED;
                            itemdone = true;
                            nelems[nest_level - 1]++;
                        } else if (!array_isspace(*ptr)) {
                            /*
                             * Other non-space characters must be after a
                             * level start, after an element start, or after
                             * an element delimiter. In any case we now must
                             * be past an element start.
                             */
                            if (parse_state != ARRAY_LEVEL_STARTED && parse_state != ARRAY_ELEM_STARTED &&
                                parse_state != ARRAY_ELEM_DELIMITED)
                                ereport(ERROR,
                                    (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
                                        errmsg("malformed array literal: \"%s\"", str)));
                            parse_state = ARRAY_ELEM_STARTED;
                        }
                    }
                    break;
            }
            if (!itemdone)
                ptr += pg_mblen(ptr);
        }
        temp[ndim - 1]++;
        ptr += pg_mblen(ptr);
    }

    /* only whitespace is allowed after the closing brace */
    while (*ptr) {
        if (!array_isspace(*ptr++))
            ereport(
                ERROR, (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION), errmsg("malformed array literal: \"%s\"", str)));
    }

    /* special case for an empty array */
    if (empty_array)
        return 0;

    for (i = 0; i < ndim; ++i)
        dim[i] = temp[i];

    return ndim;
}

/*
 * ReadArrayStr :
 *	 parses the array string pointed to by "arrayStr" and converts the values
 *	 to internal format.  Unspecified elements are initialized to nulls.
 *	 The array dimensions must already have been determined.
 *
 * Inputs:
 *	arrayStr: the string to parse.
 *			  CAUTION: the contents of "arrayStr" will be modified!
 *	origStr: the unmodified input string, used only in error messages.
 *	nitems: total number of array elements, as already determined.
 *	ndim: number of array dimensions
 *	dim[]: array axis lengths
 *	inputproc: type-specific input procedure for element datatype.
 *	typioparam, typmod: auxiliary values to pass to inputproc.
 *	typdelim: the value delimiter (type-specific).
 *	typlen, typbyval, typalign: storage parameters of element datatype.
 *
 * Outputs:
 *	values[]: filled with converted data values.
 *	nulls[]: filled with is-null markers.
 *	*hasnulls: set TRUE iff there are any null elements.
 *	*nbytes: set to total size of data area needed (including alignment
 *		padding but not including array header overhead).
 *
 * Note that values[] and nulls[] are allocated by the caller, and must have
 * nitems elements.
 */
static void ReadArrayStr(char* arrayStr, const char* origStr, int nitems, int ndim, const int* dim, FmgrInfo* inputproc,
    Oid typioparam, int32 typmod, char typdelim, int typlen, bool typbyval, char typalign, Datum* values, bool* nulls,
    bool* hasnulls, int32* nbytes)
{
    int i, nest_level = 0;
    char* srcptr = NULL;
    bool in_quotes = false;
    bool eoArray = false;
    bool hasnull = false;
    int32 totbytes;
    int indx[MAXDIM], prod[MAXDIM];
    errno_t rc = EOK;

    mda_get_prod(ndim, dim, prod);
    rc = memset_s(indx, sizeof(indx), 0, sizeof(indx));
    securec_check(rc, "\0", "\0");

    /* Initialize is-null markers to true */
    rc = memset_s(nulls, nitems * sizeof(bool), true, nitems * sizeof(bool));
    securec_check(rc, "\0", "\0");

    /*
     * We have to remove " and \ characters to create a clean item value to
     * pass to the datatype input routine.	We overwrite each item value
     * in-place within arrayStr to do this.  srcptr is the current scan point,
     * and dstptr is where we are copying to.
     *
     * We also want to suppress leading and trailing unquoted whitespace. We
     * use the leadingspace flag to suppress leading space.  Trailing space is
     * tracked by using dstendptr to point to the last significant output
     * character.
     *
     * The error checking in this routine is mostly pro-forma, since we expect
     * that ArrayCount() already validated the string.
     */
    srcptr = arrayStr;
    while (!eoArray) {
        bool itemdone = false;
        bool leadingspace = true;
        bool hasquoting = false;
        char* itemstart = NULL;
        char* dstptr = NULL;
        char* dstendptr = NULL;

        i = -1;
        itemstart = dstptr = dstendptr = srcptr;

        while (!itemdone) {
            switch (*srcptr) {
                case '\0':
                    /* Signal a premature end of the string */
                    ereport(ERROR,
                        (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
                            errmsg("malformed array literal: \"%s\"", origStr)));
                    break;
                case '\\':
                    /* Skip backslash, copy next character as-is. */
                    srcptr++;
                    if (*srcptr == '\0')
                        ereport(ERROR,
                            (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
                                errmsg("malformed array literal: \"%s\"", origStr)));
                    *dstptr++ = *srcptr++;
                    /* Treat the escaped character as non-whitespace */
                    leadingspace = false;
                    dstendptr = dstptr;
                    hasquoting = true; /* can't be a NULL marker */
                    break;
                case '\"':
                    in_quotes = !in_quotes;
                    if (in_quotes)
                        leadingspace = false;
                    else {
                        /*
                         * Advance dstendptr when we exit in_quotes; this
                         * saves having to do it in all the other in_quotes
                         * cases.
                         */
                        dstendptr = dstptr;
                    }
                    hasquoting = true; /* can't be a NULL marker */
                    srcptr++;
                    break;
                case '{':
                    if (!in_quotes) {
                        if (nest_level >= ndim)
                            ereport(ERROR,
                                (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
                                    errmsg("malformed array literal: \"%s\"", origStr)));
                        nest_level++;
                        indx[nest_level - 1] = 0;
                        srcptr++;
                    } else
                        *dstptr++ = *srcptr++;
                    break;
                case '}':
                    if (!in_quotes) {
                        if (nest_level == 0)
                            ereport(ERROR,
                                (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
                                    errmsg("malformed array literal: \"%s\"", origStr)));
                        if (i == -1)
                            i = ArrayGetOffset0(ndim, indx, prod);
                        indx[nest_level - 1] = 0;
                        nest_level--;
                        if (nest_level == 0)
                            eoArray = itemdone = true;
                        else
                            indx[nest_level - 1]++;
                        srcptr++;
                    } else
                        *dstptr++ = *srcptr++;
                    break;
                default:
                    if (in_quotes) {
                        int charlen = pg_mblen(srcptr);
                        for (int i = 0; i < charlen; i++)
                            *dstptr++ = *srcptr++;
                    }

                    else if (*srcptr == typdelim) {
                        if (i == -1)
                            i = ArrayGetOffset0(ndim, indx, prod);
                        itemdone = true;
                        indx[ndim - 1]++;
                        srcptr++;
                    } else if (array_isspace(*srcptr)) {
                        /*
                         * If leading space, drop it immediately.  Else, copy
                         * but don't advance dstendptr.
                         */
                        if (leadingspace)
                            srcptr++;
                        else
                            *dstptr++ = *srcptr++;
                    } else {
                        int charlen = pg_mblen(srcptr);
                        for (int i = 0; i < charlen; i++)
                            *dstptr++ = *srcptr++;
                        leadingspace = false;
                        dstendptr = dstptr;
                    }
                    break;
            }
        }

        Assert(dstptr < srcptr);
        *dstendptr = '\0';

        if (i < 0 || i >= nitems)
            ereport(ERROR,
                (errcode(ERRCODE_INVALID_TEXT_REPRESENTATION), errmsg("malformed array literal: \"%s\"", origStr)));

        if (u_sess->attr.attr_sql.Array_nulls && !hasquoting && pg_strcasecmp(itemstart, "NULL") == 0) {
            /* it's a NULL item */
            values[i] = InputFunctionCall(inputproc, NULL, typioparam, typmod);
            nulls[i] = true;
        } else {
            values[i] = InputFunctionCall(inputproc, itemstart, typioparam, typmod);
            nulls[i] = false;
        }
    }

    /*
     * Check for nulls, compute total data space needed
     */
    hasnull = false;
    totbytes = 0;
    for (i = 0; i < nitems; i++) {
        if (nulls[i]) {
            hasnull = true;
        } else {
            /* let's just make sure data is not toasted */
            if (typlen == -1)
                values[i] = PointerGetDatum(PG_DETOAST_DATUM(values[i]));
            totbytes = att_addlength_datum(totbytes, typlen, values[i]);
            totbytes = att_align_nominal(totbytes, typalign);
            /* check for overflow of total request */
            if (!AllocSizeIsValid(totbytes))
                ereport(ERROR,
                    (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                        errmsg("array size exceeds the maximum allowed (%d)", (int)MaxAllocSize)));
        }
    }
    *hasnulls = hasnull;
    *nbytes = totbytes;
}

/*
 * Copy data into an array object from a temporary array of Datums.
 *
 * array: array object (with header fields already filled in)
 * values: array of Datums to be copied
 * nulls: array of is-null flags (can be NULL if no nulls)
 * nitems: number of Datums to be copied
 * typbyval, typlen, typalign: info about element datatype
 * freedata: if TRUE and element type is pass-by-ref, pfree data values
 * referenced by Datums after copying them.
 *
 * If the input data is of varlena type, the caller must have ensured that
 * the values are not toasted.	(Doing it here doesn't work since the
 * caller has already allocated space for the array...)
 */
void CopyArrayEls(ArrayType* array, Datum* values, const bool* nulls, int nitems, int typlen, bool typbyval,
    char typalign, bool freedata)
{
    char* p = ARR_DATA_PTR(array);
    bits8* bitmap = ARR_NULLBITMAP(array);
    int bitval = 0;
    uint32 bitmask = 1;
    int i;

    if (typbyval)
        freedata = false;

    for (i = 0; i < nitems; i++) {
        if (nulls != NULL && nulls[i]) {
            if (bitmap == NULL) /* shouldn't happen */
                ereport(
                    ERROR, (errcode(ERRCODE_UNEXPECTED_NULL_VALUE), errmsg("null array element where not supported")));
            /* bitmap bit stays 0 */
        } else {
            bitval |= bitmask;
            p += ArrayCastAndSet(values[i], typlen, typbyval, typalign, p);
            if (freedata)
                pfree(DatumGetPointer(values[i]));
        }
        if (bitmap != NULL) {
            bitmask <<= 1;
            if (bitmask == 0x100) {
                *bitmap++ = bitval;
                bitval = 0;
                bitmask = 1;
            }
        }
    }

    if (bitmap != NULL && bitmask != 1)
        *bitmap = bitval;
}


Datum tdigest_in(PG_FUNCTION_ARGS)
{
    // get string from dn and returns the TdigestData to cn
    char* str = PG_GETARG_CSTRING(0);

    if (NULL == strstr(str, "TdigestData")) {
        ereport(ERROR, (errmodule(MOD_OPT_AGG), errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                        errmsg("This input type is not supported for tdigest_in()"), errdetail("N/A"),
                        errcause("input type is not supported"),
                        erraction("Check tdigest_in syntax to obtain the supported privilege types")));
        PG_RETURN_NULL();
    }
    TdigestData* oldres = (TdigestData*) stringToNode(str);
    Size BuffSize = sizeof(TdigestData) + (oldres->cap * sizeof(CentroidPoint));
    SET_VARSIZE(oldres, BuffSize);
    PG_RETURN_POINTER(oldres);
}

Datum tdigest_out(PG_FUNCTION_ARGS)
{
    // get TdigestData and returns the string to cn
    TdigestData* oldres = (TdigestData*)PG_DETOAST_DATUM(PG_GETARG_DATUM(0));

    // convert TdigestData to a string
    StringInfoData str;
    initStringInfo(&str);
    appendStringInfoChar(&str, '{');
    appendStringInfoString(&str, "TdigestData");
    appendStringInfo(&str, " :" CppAsString(compression) " " "%.2f", oldres->compression);
    appendStringInfo(&str, " :" CppAsString(cap) " %d", oldres->cap);
    appendStringInfo(&str, " :" CppAsString(merged_nodes) " %d", oldres->merged_nodes);
    appendStringInfo(&str, " :" CppAsString(unmerged_nodes) " %d", oldres->unmerged_nodes);
    appendStringInfo(&str, " :" CppAsString(merged_count) " " "%.2f", oldres->merged_count);
    appendStringInfo(&str, " :" CppAsString(unmerged_count) " " "%.2f", oldres->unmerged_count);
    appendStringInfo(&str, " :" CppAsString(valuetoc) " " "%f", oldres->valuetoc);
    for (int i = 0; i < oldres->merged_nodes + oldres->unmerged_nodes; i++) {
        appendStringInfo(&str, " :nodes[%d]", i);
        appendStringInfo(&str, " %f", oldres->nodes[i].mean);
        appendStringInfo(&str, " %ld", oldres->nodes[i].count);
    }
    appendStringInfoChar(&str, '}');
    
    PG_RETURN_CSTRING(str.data);
}

/*
 * array_out :
 *		   takes the internal representation of an array and returns a string
 *		  containing the array in its external format.
 */
Datum array_out(PG_FUNCTION_ARGS)
{
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    Oid element_type = ARR_ELEMTYPE(v);
    int typlen;
    int retval_len;
    bool typbyval = false;
    char typalign;
    char typdelim;
    char *p = NULL, *tmp = NULL, *retval = NULL, **values = NULL, dims_str[(MAXDIM * 33) + 2];
    int len = (MAXDIM * 33) + 2;
    int len_p = 0;

    /*
     * 33 per dim since we assume 15 digits per number + ':' +'[]'
     *
     * +2 allows for assignment operator + trailing null
     */
    bits8* bitmap = NULL;
    uint32 bitmask;
    bool *needquotes = NULL, needdims = false;
    int nitems, overall_length, i, j, k, indx[MAXDIM];
    int ndim, *dims = NULL, *lb = NULL;
    ArrayMetaState* my_extra = NULL;
    errno_t rc = EOK;

    /*
     * We arrange to look up info about element type, including its output
     * conversion proc, only once per series of calls, assuming the element
     * type doesn't change underneath us.
     */
    my_extra = (ArrayMetaState*)fcinfo->flinfo->fn_extra;
    if (my_extra == NULL) {
        fcinfo->flinfo->fn_extra = MemoryContextAlloc(fcinfo->flinfo->fn_mcxt, sizeof(ArrayMetaState));
        my_extra = (ArrayMetaState*)fcinfo->flinfo->fn_extra;
        my_extra->element_type = ~element_type;
    }

    if (my_extra->element_type != element_type) {
        /*
         * Get info about element type, including its output conversion proc
         */
        get_type_io_data(element_type,
            IOFunc_output,
            &my_extra->typlen,
            &my_extra->typbyval,
            &my_extra->typalign,
            &my_extra->typdelim,
            &my_extra->typioparam,
            &my_extra->typiofunc);
        fmgr_info_cxt(my_extra->typiofunc, &my_extra->proc, fcinfo->flinfo->fn_mcxt);
        my_extra->element_type = element_type;
    }
    typlen = my_extra->typlen;
    typbyval = my_extra->typbyval;
    typalign = my_extra->typalign;
    typdelim = my_extra->typdelim;

    ndim = ARR_NDIM(v);
    dims = ARR_DIMS(v);
    lb = ARR_LBOUND(v);
    nitems = ArrayGetNItems(ndim, dims);

    if (nitems == 0) {
        retval = pstrdup("{}");

        /* free memory if allocated by the toaster */
        PG_FREE_IF_COPY(v, 0);

        PG_RETURN_CSTRING(retval);
    }

    /*
     * we will need to add explicit dimensions if any dimension has a lower
     * bound other than one
     */
    for (i = 0; i < ndim; i++) {
        if (lb[i] != 1) {
            needdims = true;
            break;
        }
    }

    /*
     * Convert all values to string form, count total space needed (including
     * any overhead such as escaping backslashes), and detect whether each
     * item needs double quotes.
     */
    values = (char**)palloc(nitems * sizeof(char*));
    needquotes = (bool*)palloc(nitems * sizeof(bool));
    overall_length = 1; /* don't forget to count \0 at end. */

    p = ARR_DATA_PTR(v);
    bitmap = ARR_NULLBITMAP(v);
    bitmask = 1;

    for (i = 0; i < nitems; i++) {
        bool needquote = false;

        /* Get source element, checking for NULL */
        if (bitmap && (*bitmap & bitmask) == 0) {
            values[i] = pstrdup("NULL");
            overall_length += 4;
            needquote = false;
        } else {
            Datum itemvalue;
            int charlen;

            itemvalue = fetch_att(p, typbyval, typlen);
            values[i] = OutputFunctionCall(&my_extra->proc, itemvalue);
            p = att_addlength_pointer(p, typlen, p);
            p = (char*)att_align_nominal(p, typalign);

            /* count data plus backslashes; detect chars needing quotes */
            if (values[i][0] == '\0')
                needquote = true; /* force quotes for empty string */
            else if (pg_strcasecmp(values[i], "NULL") == 0)
                needquote = true; /* force quotes for literal NULL */
            else
                needquote = false;

            tmp = values[i];
            while (*tmp != '\0') {
                char ch = *tmp;

                charlen = pg_mblen(tmp);
                overall_length += charlen;
                if (ch == '"' || ch == '\\') {
                    needquote = true;
                    overall_length += 1;
                } else if (ch == '{' || ch == '}' || ch == typdelim || array_isspace(ch))
                    needquote = true;
                tmp += charlen;
            }
        }

        needquotes[i] = needquote;

        /* Count the pair of double quotes, if needed */
        if (needquote)
            overall_length += 2;
        /* and the comma */
        overall_length += 1;

        /* advance bitmap pointer if any */
        if (bitmap != NULL) {
            bitmask <<= 1;
            if (bitmask == 0x100) {
                bitmap++;
                bitmask = 1;
            }
        }
    }

    /*
     * count total number of curly braces in output string
     */
    for (i = j = 0, k = 1; i < ndim; i++) {	    
        k *= dims[i];
        j += k;
    }

    dims_str[0] = '\0';

    /* add explicit dimensions if required */
    if (needdims) {
        char* ptr = dims_str;

        for (i = 0; i < ndim; i++) {
            rc = sprintf_s(ptr, len, "[%d:%d]", lb[i], lb[i] + dims[i] - 1);
            securec_check_ss(rc, "", "");
            len -= strlen(ptr);
            ptr += strlen(ptr);
        }
        *ptr++ = *ASSGN;
        *ptr = '\0';
    }
    retval_len = strlen(dims_str) + overall_length + 2 * j;
    retval = (char*)palloc(retval_len);
    p = retval;

#define APPENDCHAR(ch) (*p++ = (ch), *p = '\0')

    if (needdims != false) {
        rc = strcpy_s(p, retval_len, dims_str);
        securec_check(rc, "", "");
        len_p = strlen(p);
        p += len_p;
        retval_len -= len_p;
    }
    APPENDCHAR('{');
    for (i = 0; i < ndim; i++)
        indx[i] = 0;
    j = 0;
    k = 0;
    do {
        for (i = j; i < ndim - 1; i++)
            APPENDCHAR('{');

        if (needquotes[k]) {
            APPENDCHAR('"');
            tmp = values[k];
            int charlen;
            char ch;
            while (*tmp != '\0') {
                ch = *tmp;
                charlen = pg_mblen(tmp);

                if (ch == '"' || ch == '\\')
                    *p++ = '\\';
                for (int cp = 0; cp < charlen; cp++) {
                    *p++ = *tmp++;
                }
            }
            *p = '\0';
            APPENDCHAR('"');
        } else {
            rc = strcpy_s(p, retval_len, values[k]);
            securec_check(rc, "", "");
            len_p = strlen(p);
            p += len_p;
            retval_len -= len_p;
        }
        pfree(values[k++]);

        for (i = ndim - 1; i >= 0; i--) {
            indx[i] = (indx[i] + 1) % dims[i];
            if (indx[i]) {
                APPENDCHAR(typdelim);
                break;
            } else
                APPENDCHAR('}');
        }
        j = i;
    } while (j != -1);

#undef APPENDCHAR

    pfree(values);
    pfree(needquotes);

    /* free memory if allocated by the toaster */
    PG_FREE_IF_COPY(v, 0);

    PG_RETURN_CSTRING(retval);
}

/*
 * array_recv :
 *		  converts an array from the external binary format to
 *		  its internal format.
 *
 * return value :
 *		  the internal representation of the input array
 */
Datum array_recv(PG_FUNCTION_ARGS)
{
    StringInfo buf = (StringInfo)PG_GETARG_POINTER(0);
    Oid spec_element_type = PG_GETARG_OID(1); /* type of an array
                                               * element */
    int32 typmod = PG_GETARG_INT32(2);        /* typmod for array elements */
    Oid element_type;
    int typlen;
    bool typbyval = false;
    char typalign;
    Oid typioparam;
    int i, nitems;
    Datum* dataPtr = NULL;
    bool* nullsPtr = NULL;
    bool hasnulls = false;
    int32 nbytes;
    int32 dataoffset;
    ArrayType* retval = NULL;
    int ndim, flags, dim[MAXDIM], lBound[MAXDIM];
    ArrayMetaState* my_extra = NULL;
    errno_t rc = EOK;

    /* Get the array header information */
    ndim = pq_getmsgint(buf, 4);
    if (ndim < 0) /* we do allow zero-dimension arrays */
        ereport(
            ERROR, (errcode(ERRCODE_INVALID_BINARY_REPRESENTATION), errmsg("invalid number of dimensions: %d", ndim)));
    if (ndim > MAXDIM)
        ereport(ERROR,
            (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                errmsg("number of array dimensions (%d) exceeds the maximum allowed (%d)", ndim, MAXDIM)));

    flags = pq_getmsgint(buf, 4);
    if (flags != 0 && flags != 1)
        ereport(ERROR, (errcode(ERRCODE_INVALID_BINARY_REPRESENTATION), errmsg("invalid array flags")));

    element_type = pq_getmsgint(buf, sizeof(Oid));
    if (element_type != spec_element_type) {
        /* XXX Can we allow taking the input element type in any cases? */
        ereport(ERROR, (errcode(ERRCODE_DATATYPE_MISMATCH), errmsg("wrong element type")));
    }

    for (i = 0; i < ndim; i++) {
        dim[i] = pq_getmsgint(buf, 4);
        lBound[i] = pq_getmsgint(buf, 4);

        if (dim[i] < 0) {
            ereport(ERROR,
                (errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE),
                    errmsg("array size cannot be negative.")));
        }
    }

    /* This checks for overflow of array dimensions */
    nitems = ArrayGetNItems(ndim, dim);
    ArrayCheckBounds(ndim, dim, lBound);

    /*
     * We arrange to look up info about element type, including its receive
     * conversion proc, only once per series of calls, assuming the element
     * type doesn't change underneath us.
     */
    my_extra = (ArrayMetaState*)fcinfo->flinfo->fn_extra;
    if (my_extra == NULL) {
        fcinfo->flinfo->fn_extra = MemoryContextAlloc(fcinfo->flinfo->fn_mcxt, sizeof(ArrayMetaState));
        my_extra = (ArrayMetaState*)fcinfo->flinfo->fn_extra;
        my_extra->element_type = ~element_type;
    }

    if (my_extra->element_type != element_type) {
        /* Get info about element type, including its receive proc */
        get_type_io_data(element_type,
            IOFunc_receive,
            &my_extra->typlen,
            &my_extra->typbyval,
            &my_extra->typalign,
            &my_extra->typdelim,
            &my_extra->typioparam,
            &my_extra->typiofunc);
        if (!OidIsValid(my_extra->typiofunc))
            ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_FUNCTION),
                    errmsg("no binary input function available for type %s", format_type_be(element_type))));
        fmgr_info_cxt(my_extra->typiofunc, &my_extra->proc, fcinfo->flinfo->fn_mcxt);
        my_extra->element_type = element_type;
    }

    if (nitems == 0) {
        /* Return empty array ... but not till we've validated element_type */
        PG_RETURN_ARRAYTYPE_P(construct_empty_array(element_type));
    }

    typlen = my_extra->typlen;
    typbyval = my_extra->typbyval;
    typalign = my_extra->typalign;
    typioparam = my_extra->typioparam;

    dataPtr = (Datum*)palloc(nitems * sizeof(Datum));
    nullsPtr = (bool*)palloc(nitems * sizeof(bool));
    ReadArrayBinary(buf,
        nitems,
        &my_extra->proc,
        typioparam,
        typmod,
        typlen,
        typbyval,
        typalign,
        dataPtr,
        nullsPtr,
        &hasnulls,
        &nbytes);
    if (hasnulls) {
        dataoffset = ARR_OVERHEAD_WITHNULLS(ndim, nitems);
        nbytes += dataoffset;
    } else {
        dataoffset = 0; /* marker for no null bitmap */
        nbytes += ARR_OVERHEAD_NONULLS(ndim);
    }
    retval = (ArrayType*)palloc0(nbytes);
    SET_VARSIZE(retval, nbytes);
    retval->ndim = ndim;
    retval->dataoffset = dataoffset;
    retval->elemtype = element_type;
    rc = memcpy_s(ARR_DIMS(retval), ndim * sizeof(int), dim, ndim * sizeof(int));
    securec_check(rc, "\0", "\0");
    rc = memcpy_s(ARR_LBOUND(retval), ndim * sizeof(int), lBound, ndim * sizeof(int));
    securec_check(rc, "\0", "\0");

    CopyArrayEls(retval, dataPtr, nullsPtr, nitems, typlen, typbyval, typalign, true);

    pfree(dataPtr);
    pfree(nullsPtr);

    PG_RETURN_ARRAYTYPE_P(retval);
}

/*
 * ReadArrayBinary:
 *	 collect the data elements of an array being read in binary style.
 *
 * Inputs:
 *	buf: the data buffer to read from.
 *	nitems: total number of array elements (already read).
 *	receiveproc: type-specific receive procedure for element datatype.
 *	typioparam, typmod: auxiliary values to pass to receiveproc.
 *	typlen, typbyval, typalign: storage parameters of element datatype.
 *
 * Outputs:
 *	values[]: filled with converted data values.
 *	nulls[]: filled with is-null markers.
 *	*hasnulls: set TRUE iff there are any null elements.
 *	*nbytes: set to total size of data area needed (including alignment
 *		padding but not including array header overhead).
 *
 * Note that values[] and nulls[] are allocated by the caller, and must have
 * nitems elements.
 */
static void ReadArrayBinary(StringInfo buf, int nitems, FmgrInfo* receiveproc, Oid typioparam, int32 typmod, int typlen,
    bool typbyval, char typalign, Datum* values, bool* nulls, bool* hasnulls, int32* nbytes)
{
    int i;
    bool hasnull = false;
    int32 totbytes;

    for (i = 0; i < nitems; i++) {
        int itemlen;
        StringInfoData elem_buf;
        char csave;

        /* Get and check the item length */
        itemlen = pq_getmsgint(buf, 4);
        if (itemlen < -1 || itemlen > (buf->len - buf->cursor))
            ereport(
                ERROR, (errcode(ERRCODE_INVALID_BINARY_REPRESENTATION), errmsg("insufficient data left in message")));

        if (itemlen == -1) {
            /* -1 length means NULL */
            values[i] = ReceiveFunctionCall(receiveproc, NULL, typioparam, typmod);
            nulls[i] = true;
            continue;
        }

        /*
         * Rather than copying data around, we just set up a phony StringInfo
         * pointing to the correct portion of the input buffer. We assume we
         * can scribble on the input buffer so as to maintain the convention
         * that StringInfos have a trailing null.
         */
        elem_buf.data = &buf->data[buf->cursor];
        elem_buf.maxlen = itemlen + 1;
        elem_buf.len = itemlen;
        elem_buf.cursor = 0;

        buf->cursor += itemlen;

        csave = buf->data[buf->cursor];
        buf->data[buf->cursor] = '\0';

        /* Now call the element's receiveproc */
        values[i] = ReceiveFunctionCall(receiveproc, &elem_buf, typioparam, typmod);
        nulls[i] = false;

        /* Trouble if it didn't eat the whole buffer */
        if (elem_buf.cursor != itemlen)
            ereport(ERROR, (errcode(ERRCODE_INVALID_BINARY_REPRESENTATION),
                    errmsg("improper binary format in array element %d", i + 1)));

        buf->data[buf->cursor] = csave;
    }

    /*
     * Check for nulls, compute total data space needed
     */
    hasnull = false;
    totbytes = 0;
    for (i = 0; i < nitems; i++) {
        if (nulls[i])
            hasnull = true;
        else {
            /* let's just make sure data is not toasted */
            if (typlen == -1)
                values[i] = PointerGetDatum(PG_DETOAST_DATUM(values[i]));
            totbytes = att_addlength_datum(totbytes, typlen, values[i]);
            totbytes = att_align_nominal(totbytes, typalign);
            /* check for overflow of total request */
            if (!AllocSizeIsValid(totbytes))
                ereport(ERROR, (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                        errmsg("array size exceeds the maximum allowed (%d)", (int)MaxAllocSize)));
        }
    }
    *hasnulls = hasnull;
    *nbytes = totbytes;
}

/*
 * array_send :
 *		  takes the internal representation of an array and returns a bytea
 *		  containing the array in its external binary format.
 */
Datum array_send(PG_FUNCTION_ARGS)
{
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    Oid element_type = ARR_ELEMTYPE(v);
    int typlen;
    bool typbyval = false;
    char typalign;
    char* p = NULL;
    bits8* bitmap = NULL;
    uint32 bitmask;
    int nitems, i;
    int ndim, *dim = NULL;
    StringInfoData buf;
    ArrayMetaState* my_extra = NULL;

    /*
     * We arrange to look up info about element type, including its send
     * conversion proc, only once per series of calls, assuming the element
     * type doesn't change underneath us.
     */
    my_extra = (ArrayMetaState*)fcinfo->flinfo->fn_extra;
    if (my_extra == NULL) {
        fcinfo->flinfo->fn_extra = MemoryContextAlloc(fcinfo->flinfo->fn_mcxt, sizeof(ArrayMetaState));
        my_extra = (ArrayMetaState*)fcinfo->flinfo->fn_extra;
        my_extra->element_type = ~element_type;
    }

    if (my_extra->element_type != element_type) {
        /* Get info about element type, including its send proc */
        get_type_io_data(element_type,
            IOFunc_send,
            &my_extra->typlen,
            &my_extra->typbyval,
            &my_extra->typalign,
            &my_extra->typdelim,
            &my_extra->typioparam,
            &my_extra->typiofunc);
        if (!OidIsValid(my_extra->typiofunc))
            ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_FUNCTION),
                    errmsg("no binary output function available for type %s", format_type_be(element_type))));
        fmgr_info_cxt(my_extra->typiofunc, &my_extra->proc, fcinfo->flinfo->fn_mcxt);
        my_extra->element_type = element_type;
    }
    typlen = my_extra->typlen;
    typbyval = my_extra->typbyval;
    typalign = my_extra->typalign;

    ndim = ARR_NDIM(v);
    dim = ARR_DIMS(v);
    nitems = ArrayGetNItems(ndim, dim);

    pq_begintypsend(&buf);

    /* Send the array header information */
    pq_sendint32(&buf, ndim);
    pq_sendint32(&buf, ARR_HASNULL(v) ? 1 : 0);
    pq_sendint32(&buf, element_type);
    for (i = 0; i < ndim; i++) {
        pq_sendint32(&buf, ARR_DIMS(v)[i]);
        pq_sendint32(&buf, ARR_LBOUND(v)[i]);
    }

    /* Send the array elements using the element's own sendproc */
    p = ARR_DATA_PTR(v);
    bitmap = ARR_NULLBITMAP(v);
    bitmask = 1;

    for (i = 0; i < nitems; i++) {
        /* Get source element, checking for NULL */
        if (bitmap && (*bitmap & bitmask) == 0) {
            /* -1 length means a NULL */
            pq_sendint32(&buf, -1);
        } else {
            Datum itemvalue;
            bytea* outputbytes = NULL;

            itemvalue = fetch_att(p, typbyval, typlen);
            outputbytes = SendFunctionCall(&my_extra->proc, itemvalue);
            pq_sendint32(&buf, VARSIZE(outputbytes) - VARHDRSZ);
            pq_sendbytes(&buf, VARDATA(outputbytes), VARSIZE(outputbytes) - VARHDRSZ);
            pfree(outputbytes);

            p = att_addlength_pointer(p, typlen, p);
            p = (char*)att_align_nominal(p, typalign);
        }

        /* advance bitmap pointer if any */
        if (bitmap != NULL) {
            bitmask <<= 1;
            if (bitmask == 0x100) {
                bitmap++;
                bitmask = 1;
            }
        }
    }

    PG_RETURN_BYTEA_P(pq_endtypsend(&buf));
}

/*
 * array_ndims :
 *		  returns the number of dimensions of the array pointed to by "v"
 */
Datum array_ndims(PG_FUNCTION_ARGS)
{
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);

    /* Sanity check: does it look like an array at all? */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM)
        PG_RETURN_NULL();

    PG_RETURN_INT32(ARR_NDIM(v));
}

/*
 * array_dims :
 *		  returns the dimensions of the array pointed to by "v", as a "text"
 */
Datum array_dims(PG_FUNCTION_ARGS)
{
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    char* p = NULL;
    int i;
    int *dimv = NULL, *lb = NULL;
    errno_t rc = EOK;

    /*
     * 33 since we assume 15 digits per number + ':' +'[]'
     *
     * +1 for trailing null
     */
    int len = MAXDIM * 33 + 1;
    char buf[MAXDIM * 33 + 1];

    /* Sanity check: does it look like an array at all? */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM)
        PG_RETURN_NULL();

    dimv = ARR_DIMS(v);
    lb = ARR_LBOUND(v);

    p = buf;
    for (i = 0; i < ARR_NDIM(v); i++) {
        rc = sprintf_s(p, len, "[%d:%d]", lb[i], dimv[i] + lb[i] - 1);
        securec_check_ss(rc, "", "");
        len -= strlen(p);
        p += strlen(p);
    }

    PG_RETURN_TEXT_P(cstring_to_text(buf));
}

/*
 * array_lower :
 *		returns the lower dimension, of the DIM requested, for
 *		the array pointed to by "v", as an int4
 */
Datum array_lower(PG_FUNCTION_ARGS)
{
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    int reqdim = PG_GETARG_INT32(1);
    int *lb = NULL;
    int result;

    /* Sanity check: does it look like an array at all */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        PG_RETURN_NULL();
    }

    /* Sanity check: was the requested dim valid */
    if (reqdim <= 0 || reqdim > ARR_NDIM(v)) {
        PG_RETURN_NULL();
    }

    lb = ARR_LBOUND(v);
    result = lb[reqdim - 1];

    PG_RETURN_INT32(result);
}

/*
 * array_upper :
 *		returns the upper dimension, of the DIM requested, for
 *		the array pointed to by "v", as an int4
 */
Datum array_upper(PG_FUNCTION_ARGS)
{
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    int reqdim = PG_GETARG_INT32(1);
    int *dimv = NULL;
    int *lb = NULL;
    int result;

    /* Sanity check: does it look like an array at all? */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        PG_RETURN_NULL();
    }

    /* Sanity check: was the requested dim valid */
    if (reqdim <= 0 || reqdim > ARR_NDIM(v)) {
        PG_RETURN_NULL();
    }

    lb = ARR_LBOUND(v);
    dimv = ARR_DIMS(v);

    result = dimv[reqdim - 1] + lb[reqdim - 1] - 1;

    PG_RETURN_INT32(result);
}

/*
 * array_length :
 *		returns the length, of the dimension requested, for
 *		the array pointed to by "v", as an int4
 */
Datum array_length(PG_FUNCTION_ARGS)
{
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    int reqdim = PG_GETARG_INT32(1);
    int* dimv = NULL;
    int result = 0;

    /* Sanity check: does it look like an array at all */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        if (u_sess->attr.attr_sql.sql_compatibility != A_FORMAT) {
            PG_RETURN_NULL();
        } else {
            PG_RETURN_INT32(result);
        }
    }

    /* Sanity check: was the requested dim valid */
    if (reqdim <= 0 || reqdim > ARR_NDIM(v)) {
        if (u_sess->attr.attr_sql.sql_compatibility != A_FORMAT) {
            PG_RETURN_NULL();
        } else {
            PG_RETURN_INT32(result);
        }
    }

    dimv = ARR_DIMS(v);

    result = dimv[reqdim - 1];

    PG_RETURN_INT32(result);
}

Datum array_indexby_length(PG_FUNCTION_ARGS)
{
    checkEnv();
    if (PG_ARGISNULL(0)) {
        PG_RETURN_INT32(0);
    }
    
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    int reqdim = PG_GETARG_INT32(1);
    int* dimv = NULL;
    int result = 0;

    /* Sanity check: does it look like an array at all */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        if (u_sess->attr.attr_sql.sql_compatibility != A_FORMAT) {
            PG_RETURN_NULL();
        } else {
            PG_RETURN_INT32(result);
        }
    }

    /* Sanity check: was the requested dim valid */
    if (reqdim <= 0 || reqdim > ARR_NDIM(v)) {
        if (u_sess->attr.attr_sql.sql_compatibility != A_FORMAT) {
            PG_RETURN_NULL();
        } else {
            PG_RETURN_INT32(result);
        }
    }

    dimv = ARR_DIMS(v);

    result = dimv[reqdim - 1];

    PG_RETURN_INT32(result);
}

/*
 * array_exists:
 *    returns whether index element is null
 */
Datum array_exists(PG_FUNCTION_ARGS)
{
    if (PG_ARGISNULL(0) || PG_ARGISNULL(1)) {
        PG_RETURN_BOOL(false);
    }
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    int index = PG_GETARG_INT32(1);
    int* dimv = NULL;
    int length = 0;
    int *lb = NULL;
    int lower;
    int upper;
    bool result = true;

    /* Sanity check: does it look like an array at all */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        result = false;
        PG_RETURN_BOOL(result);
    }

    dimv = ARR_DIMS(v);
    length = dimv[0];
    lb = ARR_LBOUND(v);
    lower = lb[0];
    upper = lower + length - 1;

    /* if index is not in [lower, upper] range, return false */
    if (index < lower || index > upper) {
        result = false;
    }
    PG_RETURN_BOOL(result);
}

static void checkEnv()
{
#ifdef ENABLE_MULTIPLE_NODES
    ereport(ERROR, (errcode(ERRCODE_WRONG_OBJECT_TYPE),
            errmsg("distribute database not support this function")));
#endif
}

static bool array_index_exists_internal(ArrayType* v, HTAB* table_index, Oid tableOfIndexType, Datum index_datum)
{
    TableOfIndexKey key;
    key.exprtypeid = tableOfIndexType;
    key.exprdatum = index_datum;
    int index = getTableOfIndexByDatumValue(key, table_index, NULL);

    int* dimv = NULL;
    int length = 0;
    int *lb = NULL;
    int lower;
    int upper;
    bool result = true;
    dimv = ARR_DIMS(v);
    length = dimv[0];
    lb = ARR_LBOUND(v);
    lower = lb[0];
    upper = lower + length - 1;

    /* if index is not in [lower, upper] range, return false */
    if (index < lower || index > upper) {
        result = false;
    }
    return result;
}

/*
 * array_varchar_exists:
 *    returns whether index element is null
 */
Datum array_varchar_exists(PG_FUNCTION_ARGS)
{
    checkEnv();
    if (PG_ARGISNULL(0) || PG_ARGISNULL(1)) {
        PG_RETURN_BOOL(false);
    }
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    /* Sanity check: does it look like an array at all */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        PG_RETURN_BOOL(false);
    }
    Datum index_datum = PG_GETARG_DATUM(1);
    if (u_sess->SPI_cxt.cur_tableof_index == NULL || 
        u_sess->SPI_cxt.cur_tableof_index->tableOfIndex == NULL) {
        ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
            errmsg("array_varchar_exists cannot be executed when tableOfIndex is not initialised.")));
    }

    /* transfer varchar format */
    bool isTran = false;
    if (VARATT_IS_1B(index_datum)) {
        index_datum = transVaratt1BTo4B(index_datum);
        isTran = true;
    }

    bool result = array_index_exists_internal(v,
                                              u_sess->SPI_cxt.cur_tableof_index->tableOfIndex,
                                              u_sess->SPI_cxt.cur_tableof_index->tableOfIndexType,
                                              index_datum);
    if (isTran) {
        pfree(DatumGetPointer(index_datum));
    }
    PG_RETURN_BOOL(result);
}

Datum array_integer_exists(PG_FUNCTION_ARGS)
{
    checkEnv();
    if (PG_ARGISNULL(0) || PG_ARGISNULL(1)) {
        PG_RETURN_BOOL(false);
    }
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    /* Sanity check: does it look like an array at all */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        PG_RETURN_BOOL(false);
    }
    Datum index_datum = PG_GETARG_DATUM(1);
    if (u_sess->SPI_cxt.cur_tableof_index == NULL ||
        u_sess->SPI_cxt.cur_tableof_index->tableOfIndex == NULL) {
        ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
            errmsg("array_integer_exists cannot be executed when tableOfIndex is not initialised.")));
    }

    bool result = array_index_exists_internal(v,
                                              u_sess->SPI_cxt.cur_tableof_index->tableOfIndex,
                                              u_sess->SPI_cxt.cur_tableof_index->tableOfIndexType,
                                              index_datum);
    PG_RETURN_BOOL(result);
}

/*
 * array_next :
 *    returns next index of current index
 */
Datum array_next(PG_FUNCTION_ARGS)
{
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    int index = PG_GETARG_INT32(1);
    int* dimv = NULL;
    int length = 0;
    int *lb = NULL;
    int lower;
    int upper;
    int result = 0;

    /* Sanity check: does it look like an array at all? */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        PG_RETURN_NULL();
    }

    dimv = ARR_DIMS(v);
    length = dimv[0];
    lb = ARR_LBOUND(v);
    lower = lb[0];
    upper = lower + length - 1;

    /* if index is more than upper, return null */
    if (index >= upper) {
        PG_RETURN_NULL();
    }

    result =  (index < lower) ? lower : (index + 1);
    PG_RETURN_INT32(result);
}

static bool IsIndexGreater(Datum datum1, Datum datum2)
{
    text* arg1 = DatumGetTextPP(datum1);
    text* arg2 = DatumGetTextPP(datum2);
    /* if arg1 > arg2, return true */
    return (text_cmp(arg1, arg2, DEFAULT_COLLATION_OID) > 0);
}

static Datum tableOfIndexVarcharNextValue(HTAB* tableOfIndex, TableOfIndexKey* cmpKey)
{
    HASH_SEQ_STATUS hashSeq;
    hash_seq_init(&hashSeq, tableOfIndex);
    TableOfIndexEntry* srcEntry = NULL;
    Datum resultDatum = (Datum)0;
    while ((srcEntry = (TableOfIndexEntry*)hash_seq_search(&hashSeq)) != NULL) {
        if (IsIndexGreater(srcEntry->key.exprdatum, cmpKey->exprdatum)) {
            if (resultDatum == (Datum)0) {
                resultDatum = srcEntry->key.exprdatum;
            } else if (IsIndexGreater(resultDatum, srcEntry->key.exprdatum)) {
                resultDatum = srcEntry->key.exprdatum;
            }
        }
    }

    return resultDatum;
}

static Datum tableOfIndexVarcharPriorValue(HTAB* tableOfIndex, TableOfIndexKey* cmpKey)
{
    HASH_SEQ_STATUS hashSeq;
    hash_seq_init(&hashSeq, tableOfIndex);
    TableOfIndexEntry* srcEntry = NULL;
    Datum resultDatum = (Datum)0;
    while ((srcEntry = (TableOfIndexEntry*)hash_seq_search(&hashSeq)) != NULL) {
        if (IsIndexGreater(cmpKey->exprdatum, srcEntry->key.exprdatum)) {
            if (resultDatum == (Datum)0) {
                resultDatum = srcEntry->key.exprdatum;
            } else if (IsIndexGreater(srcEntry->key.exprdatum, resultDatum)) {
                resultDatum = srcEntry->key.exprdatum;
            }
        }
    }

    return resultDatum;
}

static Datum tableOfIndexVarcharFirstValue(HTAB* tableOfIndex)
{
    HASH_SEQ_STATUS hashSeq;
    hash_seq_init(&hashSeq, tableOfIndex);
    TableOfIndexEntry* srcEntry = NULL;
    Datum resultDatum = (Datum)0;
    while ((srcEntry = (TableOfIndexEntry*)hash_seq_search(&hashSeq)) != NULL) {
        if (resultDatum == (Datum)0) {
            resultDatum = srcEntry->key.exprdatum;
        } else if (IsIndexGreater(resultDatum, srcEntry->key.exprdatum)) {
            resultDatum = srcEntry->key.exprdatum;
        }
    }

    return resultDatum;
}

static Datum tableOfIndexVarcharLastValue(HTAB* tableOfIndex)
{
    HASH_SEQ_STATUS hashSeq;
    hash_seq_init(&hashSeq, tableOfIndex);
    TableOfIndexEntry* srcEntry = NULL;
    Datum resultDatum = (Datum)0;
    while ((srcEntry = (TableOfIndexEntry*)hash_seq_search(&hashSeq)) != NULL) {
        if (resultDatum == (Datum)0) {
            resultDatum = srcEntry->key.exprdatum;
        } else if (IsIndexGreater(srcEntry->key.exprdatum, resultDatum)) {
            resultDatum = srcEntry->key.exprdatum;
        }
    }

    return resultDatum;
}

/*
 * array_varchar_next :
 *    returns next index of current index, only for varchar type index, return varchar type of index.
 */
Datum array_varchar_next(PG_FUNCTION_ARGS)
{
    checkEnv();
    if (PG_ARGISNULL(0)) {
        PG_RETURN_NULL();
    }
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    /* Sanity check: does it look like an array at all */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        PG_RETURN_NULL();
    }

    int index = 0;
    Datum index_datum = PG_GETARG_DATUM(1);
    if (u_sess->SPI_cxt.cur_tableof_index == NULL ||
        u_sess->SPI_cxt.cur_tableof_index->tableOfIndex == NULL) {
        ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
            errmsg("array_varchar_next cannot be executed when tableOfIndex is not initialised.")));
    }

    /* transfer varchar format */
    bool isTran = false;
    if (VARATT_IS_1B(index_datum)) {
        index_datum = transVaratt1BTo4B(index_datum);
        isTran = true;
    }

    /* turn varchar index */
    HTAB* table_index = u_sess->SPI_cxt.cur_tableof_index->tableOfIndex;
    TableOfIndexKey key;
    key.exprtypeid = u_sess->SPI_cxt.cur_tableof_index->tableOfIndexType;
    key.exprdatum = index_datum;
    index = getTableOfIndexByDatumValue(key, table_index, NULL);
    /* if exist index? */
    if (index < 0) {
        if (isTran) {
            pfree(DatumGetPointer(index_datum));
        }
        PG_RETURN_NULL();
    }
    Datum next_datum = tableOfIndexVarcharNextValue(table_index, &key);
    if (isTran) {
        pfree(DatumGetPointer(index_datum));
    }
    if (next_datum == Datum(0)) {
        PG_RETURN_NULL();
    } else {
        PG_RETURN_VARCHAR_P(next_datum);
    }
}

Datum array_varchar_prior(PG_FUNCTION_ARGS)
{
    checkEnv();
    if (PG_ARGISNULL(0)) {
        PG_RETURN_NULL();
    }
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    /* Sanity check: does it look like an array at all */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        PG_RETURN_NULL();
    }
    
    int index = 0;
    Datum index_datum = PG_GETARG_DATUM(1);
    if (u_sess->SPI_cxt.cur_tableof_index == NULL ||
        u_sess->SPI_cxt.cur_tableof_index->tableOfIndex == NULL) {
        ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
            errmsg("array_varchar_prior cannot be executed when tableOfIndex is not initialised.")));
    }
    /* transfer varchar format */
    bool isTran = false;
    if (VARATT_IS_1B(index_datum)) {
        index_datum = transVaratt1BTo4B(index_datum);
        isTran = true;
    }
    /* turn varchar index */
    HTAB* table_index = u_sess->SPI_cxt.cur_tableof_index->tableOfIndex;
    TableOfIndexKey key;
    key.exprtypeid = u_sess->SPI_cxt.cur_tableof_index->tableOfIndexType;
    key.exprdatum = index_datum;
    index = getTableOfIndexByDatumValue(key, table_index, NULL);
    if (index < 0) {
        if (isTran) {
            pfree(DatumGetPointer(index_datum));
        }
        PG_RETURN_NULL();
    }
    Datum prior_datum = tableOfIndexVarcharPriorValue(table_index, &key);
    if (isTran) {
        pfree(DatumGetPointer(index_datum));
    }
    if (prior_datum == Datum(0)) {
        PG_RETURN_NULL();
    } else {
        PG_RETURN_VARCHAR_P(prior_datum);
    }
}

Datum array_varchar_first(PG_FUNCTION_ARGS)
{
    checkEnv();
    if (PG_ARGISNULL(0)) {
        PG_RETURN_NULL();
    }
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    /* Sanity check: does it look like an array at all */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        PG_RETURN_NULL();
    }
    
    if (u_sess->SPI_cxt.cur_tableof_index == NULL ||
        u_sess->SPI_cxt.cur_tableof_index->tableOfIndex == NULL) {
        ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
            errmsg("array_varchar_first cannot be executed when tableOfIndex is not initialised.")));
    }
    /* turn varchar index */
    HTAB* table_index = u_sess->SPI_cxt.cur_tableof_index->tableOfIndex;
    Datum first_datum = tableOfIndexVarcharFirstValue(table_index);
    if (first_datum == Datum(0)) {
        PG_RETURN_NULL();
    } else {
        PG_RETURN_VARCHAR_P(first_datum);
    }
}

static Datum tableOfIndexIntegerNextValue(HTAB* tableOfIndex, TableOfIndexKey* cmpKey, bool* isAssign)
{
    HASH_SEQ_STATUS hashSeq;
    hash_seq_init(&hashSeq, tableOfIndex);
    TableOfIndexEntry* srcEntry = NULL;
    Datum resultDatum = (Datum)0;
    while ((srcEntry = (TableOfIndexEntry*)hash_seq_search(&hashSeq)) != NULL) {
        if (DatumGetInt32(srcEntry->key.exprdatum) > DatumGetInt32(cmpKey->exprdatum)) {
            if (!*isAssign) {
                resultDatum = srcEntry->key.exprdatum;
                *isAssign = true;
            } else if (DatumGetInt32(resultDatum) > DatumGetInt32(srcEntry->key.exprdatum)) {
                resultDatum = srcEntry->key.exprdatum;
            }
        }
    }

    return resultDatum;
}

Datum array_integer_next(PG_FUNCTION_ARGS)
{
    checkEnv();
    if (PG_ARGISNULL(0)) {
        PG_RETURN_NULL();
    }
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    /* Sanity check: does it look like an array at all */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        PG_RETURN_NULL();
    }

    Datum index_datum = PG_GETARG_DATUM(1);
    if (u_sess->SPI_cxt.cur_tableof_index == NULL ||
        u_sess->SPI_cxt.cur_tableof_index->tableOfIndex == NULL) {
        ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
            errmsg("array_integer_next cannot be executed when tableOfIndex is not initialised.")));
    }
    /* turn integer index */
    HTAB* table_index = u_sess->SPI_cxt.cur_tableof_index->tableOfIndex;
    TableOfIndexKey key;
    key.exprtypeid = u_sess->SPI_cxt.cur_tableof_index->tableOfIndexType;
    key.exprdatum = index_datum;
    int index = getTableOfIndexByDatumValue(key, table_index, NULL);
    /* if exist index? */
    if (index < 0) {
        PG_RETURN_NULL();
    }
    bool isAssign = false;
    Datum next_datum = tableOfIndexIntegerNextValue(table_index, &key, &isAssign);

    if (isAssign) {
        PG_RETURN_INT32(next_datum);
    } else {
        PG_RETURN_NULL();
    }
}

static Datum tableOfIndexIntegerPriorValue(HTAB* tableOfIndex, TableOfIndexKey* cmpKey, bool* isAssign)
{
    HASH_SEQ_STATUS hashSeq;
    hash_seq_init(&hashSeq, tableOfIndex);
    TableOfIndexEntry* srcEntry = NULL;
    Datum resultDatum = (Datum)0;
    while ((srcEntry = (TableOfIndexEntry*)hash_seq_search(&hashSeq)) != NULL) {
        if (DatumGetInt32(cmpKey->exprdatum) > DatumGetInt32(srcEntry->key.exprdatum)) {
            if (!*isAssign) {
                resultDatum = srcEntry->key.exprdatum;
                *isAssign = true;
            } else if (DatumGetInt32(srcEntry->key.exprdatum) > DatumGetInt32(resultDatum)) {
                resultDatum = srcEntry->key.exprdatum;
            }
        }
    }

    return resultDatum;
}

Datum array_integer_prior(PG_FUNCTION_ARGS)
{
    checkEnv();
    if (PG_ARGISNULL(0)) {
        PG_RETURN_NULL();
    }
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    /* Sanity check: does it look like an array at all */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        PG_RETURN_NULL();
    }
    
    Datum index_datum = PG_GETARG_DATUM(1);
    if (u_sess->SPI_cxt.cur_tableof_index == NULL ||
        u_sess->SPI_cxt.cur_tableof_index->tableOfIndex == NULL) {
        ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
            errmsg("array_integer_prior cannot be executed when tableOfIndex is not initialised.")));
    }
    /* turn varchar index */
    HTAB* table_index = u_sess->SPI_cxt.cur_tableof_index->tableOfIndex;
    TableOfIndexKey key;
    key.exprtypeid = u_sess->SPI_cxt.cur_tableof_index->tableOfIndexType;
    key.exprdatum = index_datum;
    int index = getTableOfIndexByDatumValue(key, table_index, NULL);
    if (index < 0) {
        PG_RETURN_NULL();
    }
    bool isAssign = false;
    Datum prior_datum = tableOfIndexIntegerPriorValue(table_index, &key, &isAssign);
    if (isAssign) {
        PG_RETURN_INT32(prior_datum);
    } else {
        PG_RETURN_NULL();
    }
}

static Datum tableOfIndexIntegerFirstValue(HTAB* tableOfIndex)
{
    HASH_SEQ_STATUS hashSeq;
    hash_seq_init(&hashSeq, tableOfIndex);
    TableOfIndexEntry* srcEntry = NULL;
    Datum resultDatum = (Datum)0;
    bool isAssign = false;
    while ((srcEntry = (TableOfIndexEntry*)hash_seq_search(&hashSeq)) != NULL) {
        if (!isAssign) {
            resultDatum = srcEntry->key.exprdatum;
            isAssign = true;
        } else if (DatumGetInt32(resultDatum) > DatumGetInt32(srcEntry->key.exprdatum)) {
            resultDatum = srcEntry->key.exprdatum;
        }
    }

    return resultDatum;
}

Datum array_integer_first(PG_FUNCTION_ARGS)
{
    checkEnv();
    if (PG_ARGISNULL(0)) {
        PG_RETURN_NULL();
    }
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    /* Sanity check: does it look like an array at all */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        PG_RETURN_NULL();
    }
    
    if (u_sess->SPI_cxt.cur_tableof_index == NULL ||
        u_sess->SPI_cxt.cur_tableof_index->tableOfIndex == NULL) {
        ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
            errmsg("array_integer_first cannot be executed when tableOfIndex is not initialised.")));
    }

    HTAB* table_index = u_sess->SPI_cxt.cur_tableof_index->tableOfIndex;
    if (hash_get_num_entries(table_index) == 0) {
        PG_RETURN_NULL();
    }
    Datum first_datum = tableOfIndexIntegerFirstValue(table_index);
    PG_RETURN_INT32(first_datum);
}

static Datum tableOfIndexIntegerLastValue(HTAB* tableOfIndex)
{
    HASH_SEQ_STATUS hashSeq;
    hash_seq_init(&hashSeq, tableOfIndex);
    TableOfIndexEntry* srcEntry = NULL;
    Datum resultDatum = (Datum)0;
    bool isAssign = false;
    while ((srcEntry = (TableOfIndexEntry*)hash_seq_search(&hashSeq)) != NULL) {
        if (!isAssign) {
            resultDatum = srcEntry->key.exprdatum;
            isAssign = true;
        } else if (DatumGetInt32(resultDatum) < DatumGetInt32(srcEntry->key.exprdatum)) {
            resultDatum = srcEntry->key.exprdatum;
        }
    }

    return resultDatum;
}

Datum array_integer_last(PG_FUNCTION_ARGS)
{
    checkEnv();
    if (PG_ARGISNULL(0)) {
        PG_RETURN_NULL();
    }
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    /* Sanity check: does it look like an array at all */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        PG_RETURN_NULL();
    }
    
    if (u_sess->SPI_cxt.cur_tableof_index == NULL ||
        u_sess->SPI_cxt.cur_tableof_index->tableOfIndex == NULL) {
        ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
            errmsg("array_integer_last cannot be executed when tableOfIndex is not initialised.")));
    }

    HTAB* table_index = u_sess->SPI_cxt.cur_tableof_index->tableOfIndex;
    if (hash_get_num_entries(table_index) == 0) {
        PG_RETURN_NULL();
    }
    Datum last_datum = tableOfIndexIntegerLastValue(table_index);
    PG_RETURN_INT32(last_datum);
}

Datum array_varchar_last(PG_FUNCTION_ARGS)
{
    checkEnv();
    if (PG_ARGISNULL(0)) {
        PG_RETURN_NULL();
    }
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    /* Sanity check: does it look like an array at all */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        PG_RETURN_NULL();
    }
    
    if (u_sess->SPI_cxt.cur_tableof_index == NULL ||
        u_sess->SPI_cxt.cur_tableof_index->tableOfIndex == NULL) {
        ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
            errmsg("array_varchar_last cannot be executed when tableOfIndex is not initialised.")));
    }
    /* turn varchar index */
    HTAB* table_index = u_sess->SPI_cxt.cur_tableof_index->tableOfIndex;
    Datum last_datum = tableOfIndexVarcharLastValue(table_index);
    if (last_datum == Datum(0)) {
        PG_RETURN_NULL();
    } else {
        PG_RETURN_VARCHAR_P(last_datum);
    }
}

void del_array_idx_in_hash(HTAB* hashp, int delIdx)
{
    HASH_SEQ_STATUS status;
    hash_seq_init(&status, hashp);
    TableOfIndexEntry* entry = NULL;
    while (entry = ((TableOfIndexEntry*)hash_seq_search(&status))) {
        if (delIdx > 0 && entry->index > delIdx) {
            entry->index = entry->index  - 1;
        }
    }
}

static ArrayType* array_index_delete_internal(ArrayType* v, HTAB* table_index, Oid tableOfIndexType, Datum index_datum)
{
    TableOfIndexKey key;
    key.exprtypeid = tableOfIndexType;
    key.exprdatum = index_datum;
    PLpgSQL_var* var = NULL;
    int index = getTableOfIndexByDatumValue(key, table_index, &var);
    if (index < 0) {
        return v;
    }

    ArrayType* array = array_deleteidx_internal(v, index);
    bool found = false;
    (void)hash_search(table_index, (const void*)&key, HASH_REMOVE, &found);
    del_array_idx_in_hash(table_index, index);
    /* for nest table, need delete inner vars */
    if (var != NULL && var->tableOfIndex != NULL) {
        HASH_SEQ_STATUS hashSeq;
        hash_seq_init(&hashSeq, var->tableOfIndex);
        TableOfIndexEntry* srcEntry = NULL;
        while ((srcEntry = (TableOfIndexEntry*)hash_seq_search(&hashSeq)) != NULL) {
            var->value = (Datum)array_index_delete_internal(DatumGetArrayTypeP(var->value), var->tableOfIndex,
                                                            var->tableOfIndexType, srcEntry->key.exprdatum);
        }
    }

    return array;
}

static ArrayType* array_index_delete_internal_db_a(ArrayType* v, HTAB* table_index,
		  Oid tableOfIndexType, Datum index_datum)
{
    TableOfIndexKey key;
    key.exprtypeid = tableOfIndexType;
    key.exprdatum = index_datum;
    PLpgSQL_var* var = NULL;
    ArrayType* array = NULL;
    bool found = false;
    int index = 0;
    index = getTableOfIndexByDatumValue(key, table_index, &var);
    if (index < 0) {
        return v;
    }
    array = array_deleteidx_internal_db_a(v, index, -1, false);

    (void)hash_search(table_index, (const void*)&key, HASH_REMOVE, &found);

    /* for nest table, need delete inner vars */
    if (var != NULL && var->tableOfIndex != NULL) {
        HASH_SEQ_STATUS hashSeq;
        hash_seq_init(&hashSeq, var->tableOfIndex);
        TableOfIndexEntry* srcEntry = NULL;
        while ((srcEntry = (TableOfIndexEntry*)hash_seq_search(&hashSeq)) != NULL) {
            var->value = (Datum)array_index_delete_internal_db_a(
                DatumGetArrayTypeP(var->value),
                var->tableOfIndex,
                var->tableOfIndexType,
                srcEntry->key.exprdatum);
        }
    }

    return array;
}

Datum array_integer_deleteidx(PG_FUNCTION_ARGS)
{
    checkEnv();
    if (PG_ARGISNULL(0)) {
        PG_RETURN_NULL();
    }
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    /* Sanity check: does it look like an array at all */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        PG_RETURN_ARRAYTYPE_P(v);
    }

    Datum index_datum = PG_GETARG_DATUM(1);
    if (u_sess->SPI_cxt.cur_tableof_index == NULL ||
        u_sess->SPI_cxt.cur_tableof_index->tableOfIndex == NULL) {
        int ndim = ARR_NDIM(v);
        if (ndim > 1) {
            ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                errmsg("Multidimensional arrays currently do not support DELETE (X) or DELETE (X, Y).")));
        } else {
            ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                errmsg("array_integer_deleteidx cannot be executed when tableOfIndex is not initialised.")));
        }
    }
    pthread_rwlock_wrlock(&u_sess->SPI_cxt.cur_tableof_index->tableOfIndexLock);
    ArrayType* array = array_index_delete_internal(v,
                                                   u_sess->SPI_cxt.cur_tableof_index->tableOfIndex,
                                                   u_sess->SPI_cxt.cur_tableof_index->tableOfIndexType,
                                                   index_datum);
    pthread_rwlock_unlock(&u_sess->SPI_cxt.cur_tableof_index->tableOfIndexLock);
    PG_RETURN_ARRAYTYPE_P(array);
}

static ArrayType* array_integer_deleteidx_db_a_inner(ArrayType* v, Datum index_datum)
{
    if (u_sess->SPI_cxt.cur_tableof_index == NULL ||
        u_sess->SPI_cxt.cur_tableof_index->tableOfIndex == NULL) {
        ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
            errmsg("array_integer_multi_deleteidx_db_a cannot be executed when tableOfIndex is not initialised.")));
    }

    pthread_rwlock_wrlock(&u_sess->SPI_cxt.cur_tableof_index->tableOfIndexLock);
    ArrayType* array = array_index_delete_internal_db_a(v,
                                                        u_sess->SPI_cxt.cur_tableof_index->tableOfIndex,
                                                        u_sess->SPI_cxt.cur_tableof_index->tableOfIndexType,
                                                        index_datum);
    pthread_rwlock_unlock(&u_sess->SPI_cxt.cur_tableof_index->tableOfIndexLock);
    return array;
}

Datum array_integer_deleteidx_db_a(PG_FUNCTION_ARGS)
{
    checkEnv();
    if (PG_ARGISNULL(0)) {
        PG_RETURN_NULL();
    }
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    /* Sanity check: does it look like an array at all */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        PG_RETURN_ARRAYTYPE_P(v);
    }
    Datum index_datum = PG_GETARG_DATUM(1);
    ArrayType* array = array_integer_deleteidx_db_a_inner(v, index_datum);
    PG_RETURN_ARRAYTYPE_P(array);
}

Datum array_integer_multi_deleteidx_db_a(PG_FUNCTION_ARGS)
{
    checkEnv();
    if (PG_ARGISNULL(0)) {
        PG_RETURN_NULL();
    }
    ereport(ERROR,
        (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
            errmsg("Arrays with index currently do not support DELETE (X, Y).")));
    PG_RETURN_NULL();
}

Datum array_varchar_deleteidx(PG_FUNCTION_ARGS)
{
    checkEnv();
    if (PG_ARGISNULL(0)) {
        PG_RETURN_NULL();
    }
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    /* Sanity check: does it look like an array at all */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        PG_RETURN_ARRAYTYPE_P(v);
    }

    Datum index_datum = PG_GETARG_DATUM(1);
    if (u_sess->SPI_cxt.cur_tableof_index == NULL ||
        u_sess->SPI_cxt.cur_tableof_index->tableOfIndex == NULL) {
        ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
            errmsg("array_varchar_deleteidx cannot be executed when tableOfIndex is not initialised.")));
    }
    /* transfer varchar format */
    bool isTran = false;
    if (VARATT_IS_1B(index_datum)) {
        index_datum = transVaratt1BTo4B(index_datum);
        isTran = true;
    }

    pthread_rwlock_wrlock(&u_sess->SPI_cxt.cur_tableof_index->tableOfIndexLock);
    ArrayType* array = array_index_delete_internal(v,
                                                   u_sess->SPI_cxt.cur_tableof_index->tableOfIndex,
                                                   u_sess->SPI_cxt.cur_tableof_index->tableOfIndexType,
                                                   index_datum);
    pthread_rwlock_unlock(&u_sess->SPI_cxt.cur_tableof_index->tableOfIndexLock);
    if (isTran) {
        pfree(DatumGetPointer(index_datum));
    }
    PG_RETURN_ARRAYTYPE_P(array);
}

static ArrayType* array_varchar_deleteidx_db_a_inner(ArrayType* v, Datum index_datum)
{
    bool isTran1 = false;

    if (u_sess->SPI_cxt.cur_tableof_index == NULL ||
        u_sess->SPI_cxt.cur_tableof_index->tableOfIndex == NULL) {
        ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
            errmsg("array_varchar_deleteidx_db_a cannot be executed when tableOfIndex is not initialised.")));
    }
    /* transfer varchar format */
    if (VARATT_IS_1B(index_datum)) {
        index_datum = transVaratt1BTo4B(index_datum);
        isTran1 = true;
    }
    pthread_rwlock_wrlock(&u_sess->SPI_cxt.cur_tableof_index->tableOfIndexLock);
    ArrayType* array = array_index_delete_internal_db_a(v,
                                                        u_sess->SPI_cxt.cur_tableof_index->tableOfIndex,
                                                        u_sess->SPI_cxt.cur_tableof_index->tableOfIndexType,
                                                        index_datum);
    pthread_rwlock_unlock(&u_sess->SPI_cxt.cur_tableof_index->tableOfIndexLock);
    if (isTran1) {
        pfree(DatumGetPointer(index_datum));
    }
    return array;
}

Datum array_varchar_deleteidx_db_a(PG_FUNCTION_ARGS)
{
    checkEnv();
    if (PG_ARGISNULL(0)) {
        PG_RETURN_NULL();
    }
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    /* Sanity check: does it look like an array at all */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        PG_RETURN_ARRAYTYPE_P(v);
    }
    Datum index_datum = PG_GETARG_DATUM(1);
    ArrayType* array = array_varchar_deleteidx_db_a_inner(v, index_datum);
    PG_RETURN_ARRAYTYPE_P(array);
}

Datum array_varchar_multi_deleteidx_db_a(PG_FUNCTION_ARGS)
{
    checkEnv();
    if (PG_ARGISNULL(0)) {
        PG_RETURN_NULL();
    }
    ereport(ERROR,
        (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
            errmsg("Arrays with index currently do not support DELETE (X, Y).")));
    PG_RETURN_NULL();
}

/*
 * array_prior :
 *    returns previous index of current index
 */
Datum array_prior(PG_FUNCTION_ARGS)
{
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    int index = PG_GETARG_INT32(1);
    int* dimv = NULL;
    int length = 0;
    int *lb = NULL;
    int lower;
    int upper;
    int result = 0;

    /* Sanity check: does it look like an array at all? */
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        PG_RETURN_NULL();
    }

    dimv = ARR_DIMS(v);
    length = dimv[0];
    lb = ARR_LBOUND(v);
    lower = lb[0];
    upper = lower + length - 1;

    /* if index is less than lower, return null */
    if (index <= lower) {
        PG_RETURN_NULL();
    }

    result = (index > upper) ? upper : (index - 1);
    PG_RETURN_INT32(result);
}

Datum array_extendnull(PG_FUNCTION_ARGS)
{
    ArrayType *v = PG_GETARG_ARRAYTYPE_P(0);
    int count = PG_GETARG_INT32(1);
    ArrayType *array = v;
    Oid element_type;
    Datum newelem = (Datum)0;
    int16 typlen;
    bool typbyval = false;
    char typalign;
    int indx;
    int *dimv = NULL;
    int *lb = NULL;
    int lower;
    int upper;
    int length;

    if (count <= 0 || ARR_NDIM(v) > MAXDIM) {
        PG_RETURN_ARRAYTYPE_P(array);
    }

    if (ARR_NDIM(v) <= 0) {
        lower = 0;
        upper = 0;
    } else {
        dimv = ARR_DIMS(v);
        length = dimv[0];
        lb = ARR_LBOUND(v);
        lower = lb[0];
        upper = lower + length - 1;
    }

    element_type = ARR_ELEMTYPE(v);
    get_typlenbyvalalign(element_type, &typlen, &typbyval, &typalign);

    for (int i = 1; i <= count; i++) {
        indx = upper + i;
        array = array_set(array, 1, &indx, newelem, true, -1, typlen, typbyval, typalign);
    }

    PG_RETURN_ARRAYTYPE_P(array);
}

static ArrayType* array_deleteidx_internal_db_a(ArrayType *v, int delIndex1, int delIndex2, bool multi_args)
{
    Oid element_type = InvalidOid;
    Datum newelem = (Datum)0;
    int16 typlen = 0;
    bool typbyval = false;
    char typalign = 0;
    int *dimv = NULL;
    int *lb = NULL;
    int lower = 0;
    int upper = 0;
    int length = 0;
    int i = 0;
    int drop = 0;

    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        return v;
    }

    dimv = ARR_DIMS(v);
    length = dimv[0];
    lb = ARR_LBOUND(v);
    lower = lb[0];
    upper = lower + length - 1;

    element_type = ARR_ELEMTYPE(v);
    get_typlenbyvalalign(element_type, &typlen, &typbyval, &typalign);

    for (i = lower; i <= upper; i++) {
        bool isDelIdx = multi_args ? (i >= delIndex1 && i <= delIndex2) : i == delIndex1;
        if (isDelIdx) {
            drop = i;
            v = array_set(v, 1, &drop, newelem, true, -1, typlen, typbyval, typalign);
        }
    }
    return v;
}

static ArrayType* array_deleteidx_internal(ArrayType *v, int delIndex)
{
    ArrayType *array;
    Oid element_type;
    Datum newelem = (Datum)0;
    int16 typlen;
    bool typbyval = false;
    char typalign;
    int indx;
    int *dimv = NULL;
    int *lb = NULL;
    int lower;
    int upper;
    int length;
    char* ptr = NULL;
    bits8* bitmap = NULL;
    uint32 bitmask;
    bool isnull;
    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        return v;
    }

    dimv = ARR_DIMS(v);
    length = dimv[0];
    lb = ARR_LBOUND(v);
    lower = lb[0];
    upper = lower + length - 1;

    if (delIndex < lower || delIndex > upper) {
        return v;
    }
        /* The array contains only one element. An empty array is returned. */
    if (length <= 1) {
        return construct_empty_array(ARR_ELEMTYPE(v));
    }

    int valueIndex = 0;
    int newLength = length - 1;
    bool *nulls = (bool *)palloc(newLength * sizeof(bool));
    Datum *values = (Datum *)palloc(newLength * sizeof(Datum));
    element_type = ARR_ELEMTYPE(v);
    get_typlenbyvalalign(element_type, &typlen, &typbyval, &typalign);

    ptr = ARR_DATA_PTR(v);
    bitmap = ARR_NULLBITMAP(v);
    bitmask = 1;

    for (indx = lower; indx <= upper; indx++) {
        /* Get elements, checking for NULL */
        if (bitmap && (*bitmap & bitmask) == 0) {
            isnull = true;
            newelem = (Datum)0;
        } else {
            isnull = false;
            newelem = fetch_att(ptr, typbyval, typlen);
            ptr = att_addlength_pointer(ptr, typlen, ptr);
            ptr = (char*)att_align_nominal(ptr, typalign);
        }
        bitmask <<= 1;
        if (bitmask == 0x100) {
            bitmask = 1;
            if (bitmap != NULL) {
                bitmap++;
            }
        }
        /* If index is delIndex, skip */
        if (indx == delIndex) {
            continue;
        }
        values[valueIndex] = newelem;
        nulls[valueIndex] = isnull;
        valueIndex++;
    }

    int dims[1];
    int lbs[1];
    dims[0] = newLength;
    lbs[0] = lower;
    array = construct_md_array(values, nulls, 1, dims, lbs, element_type, typlen, typbyval, typalign);
 
    pfree(values);
    pfree(nulls);
    return array;
}

Datum array_deleteidx(PG_FUNCTION_ARGS)
{
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    int delIndex = PG_GETARG_INT32(1);
    ArrayType* array = array_deleteidx_internal(v, delIndex);

    PG_RETURN_ARRAYTYPE_P(array);
}

Datum array_delete(PG_FUNCTION_ARGS)
{
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    ArrayType* array = construct_empty_array(ARR_ELEMTYPE(v));
    PG_RETURN_ARRAYTYPE_P(array);
}

Datum array_deleteidx_db_a(PG_FUNCTION_ARGS)
{
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    int delIndex = PG_GETARG_INT32(1);
    ArrayType* array = array_deleteidx_internal_db_a(v, delIndex, -1, false);

    PG_RETURN_ARRAYTYPE_P(array);
}

Datum array_multi_deleteidx_db_a(PG_FUNCTION_ARGS)
{
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    int delIndex1 = PG_GETARG_INT32(1);
    int delIndex2 = PG_GETARG_INT32(2);
    ArrayType* array = array_deleteidx_internal_db_a(v, delIndex1, delIndex2, true);

    PG_RETURN_ARRAYTYPE_P(array);
}

static void deleteTableOfIndexElement(HTAB* tableOfIndex)
{
    if (tableOfIndex == NULL) {
        return;
    }
    HASH_SEQ_STATUS hashSeq;
    hash_seq_init(&hashSeq, tableOfIndex);
    TableOfIndexEntry* srcEntry = NULL;
    bool found = false;
    while ((srcEntry = (TableOfIndexEntry*)hash_seq_search(&hashSeq)) != NULL) {
        if (srcEntry->var != NULL) {
            deleteTableOfIndexElement(srcEntry->var->tableOfIndex);
        }
        (void)hash_search(tableOfIndex, (const void*)&srcEntry->key, HASH_REMOVE, &found);
    }
}

Datum array_indexby_delete(PG_FUNCTION_ARGS)
{
    checkEnv();
    ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
    ArrayType* array = construct_empty_array(ARR_ELEMTYPE(v));
    if (u_sess->SPI_cxt.cur_tableof_index == NULL ||
        u_sess->SPI_cxt.cur_tableof_index->tableOfIndex == NULL) {
        ereport(ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
            errmsg("array_indexby_delete cannot be executed when tableOfIndex is not initialised.")));
    }
    deleteTableOfIndexElement(u_sess->SPI_cxt.cur_tableof_index->tableOfIndex);
    
    PG_RETURN_ARRAYTYPE_P(array);
}

Datum array_trim(PG_FUNCTION_ARGS)
{
    ArrayType *v = PG_GETARG_ARRAYTYPE_P(0);
    int count = PG_GETARG_INT32(1);
    ArrayType *array = v;
    Oid element_type;
    Datum newelem = (Datum)0;
    int16 typlen;
    bool typbyval = false;
    char typalign;
    int indx;
    int *dimv = NULL;
    int *lb = NULL;
    int lower;
    int upper;
    int length;
    char* ptr = NULL;
    bits8* bitmap = NULL;
    uint32 bitmask;
    bool isnull;

    if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM) {
        PG_RETURN_ARRAYTYPE_P(v);
    }

    dimv = ARR_DIMS(v);
    length = dimv[0];
    lb = ARR_LBOUND(v);
    lower = lb[0];
    upper = lower + length - 1;

    if (count <= 0) {
        PG_RETURN_ARRAYTYPE_P(array);
    }
    array = construct_empty_array(ARR_ELEMTYPE(v));
    if (count >= length) {
        PG_RETURN_ARRAYTYPE_P(array);
    }

    element_type = ARR_ELEMTYPE(v);
    get_typlenbyvalalign(element_type, &typlen, &typbyval, &typalign);

    ptr = ARR_DATA_PTR(v);
    bitmap = ARR_NULLBITMAP(v);
    bitmask = 1;

    for (indx = lower; indx <= upper - count; indx++) {
        /* Get elements, checking for NULL */
        if (bitmap && (*bitmap & bitmask) == 0) {
            isnull = true;
            newelem = (Datum)0;
        } else {
            isnull = false;
            newelem = fetch_att(ptr, typbyval, typlen);
            ptr = att_addlength_pointer(ptr, typlen, ptr);
            ptr = (char*)att_align_nominal(ptr, typalign);
        }
        array = array_set(array, 1, &indx, newelem, isnull, -1, typlen, typbyval, typalign);
        bitmask <<= 1;
        if (bitmask == 0x100) {
            bitmask = 1;
            if (bitmap != NULL) {
                bitmap++;
            }
        }
    }

    PG_RETURN_ARRAYTYPE_P(array);
}

Datum array_extend(PG_FUNCTION_ARGS)
{
    PG_RETURN_VOID();
}

/*
 * array_ref :
 *	  This routine takes an array pointer and a subscript array and returns
 *	  the referenced item as a Datum.  Note that for a pass-by-reference
 *	  datatype, the returned Datum is a pointer into the array object.
 *
 * This handles both ordinary varlena arrays and fixed-length arrays.
 *
 * Inputs:
 *	array: the array object (mustn't be NULL)
 *	nSubscripts: number of subscripts supplied
 *	indx[]: the subscript values
 *	arraytyplen: pg_type.typlen for the array type
 *	elmlen: pg_type.typlen for the array's element type
 *	elmbyval: pg_type.typbyval for the array's element type
 *	elmalign: pg_type.typalign for the array's element type
 *
 * Outputs:
 *	The return value is the element Datum.
 *	*isNull is set to indicate whether the element is NULL.
 */
Datum array_ref(ArrayType* array, int nSubscripts, const int* indx, int arraytyplen, int elmlen, bool elmbyval,
    char elmalign, bool* isNull)
{
    int i, ndim, *dim = NULL, *lb = NULL, offset, fixedDim[1], fixedLb[1];
    char *arraydataptr = NULL, *retptr = NULL;
    bits8* arraynullsptr = NULL;

    if (arraytyplen > 0) {
        if (elmlen == 0) {
            *isNull = true;
            return (Datum)0;
        }

        /*
         * fixed-length arrays -- these are assumed to be 1-d, 0-based
         */
        ndim = 1;
        fixedDim[0] = arraytyplen / elmlen;
        fixedLb[0] = 0;
        dim = fixedDim;
        lb = fixedLb;
        arraydataptr = (char*)array;
        arraynullsptr = NULL;
    } else {
        /* detoast input array if necessary */
        array = DatumGetArrayTypeP(PointerGetDatum(array));

        ndim = ARR_NDIM(array);
        dim = ARR_DIMS(array);
        lb = ARR_LBOUND(array);
        arraydataptr = ARR_DATA_PTR(array);
        arraynullsptr = ARR_NULLBITMAP(array);
    }

    /*
     * Return NULL for invalid subscript
     */
    if (ndim != nSubscripts || ndim <= 0 || ndim > MAXDIM) {
        *isNull = true;
        return (Datum)0;
    }
    for (i = 0; i < ndim; i++) {
        if (indx[i] < lb[i] || indx[i] >= (dim[i] + lb[i])) {
            *isNull = true;
            return (Datum)0;
        }
    }

    /*
     * Calculate the element number
     */
    offset = ArrayGetOffset(nSubscripts, dim, lb, indx);

    /*
     * Check for NULL array element
     */
    if (array_get_isnull(arraynullsptr, offset)) {
        *isNull = true;
        return (Datum)0;
    }

    /*
     * OK, get the element
     */
    *isNull = false;
    retptr = array_seek(arraydataptr, 0, arraynullsptr, offset, elmlen, elmbyval, elmalign);
    return ArrayCast(retptr, elmbyval, elmlen);
}

/*
 * array_get_slice :
 *		   This routine takes an array and a range of indices (upperIndex and
 *		   lowerIndx), creates a new array structure for the referred elements
 *		   and returns a pointer to it.
 *
 * This handles both ordinary varlena arrays and fixed-length arrays.
 *
 * Inputs:
 *	array: the array object (mustn't be NULL)
 *	nSubscripts: number of subscripts supplied (must be same for upper/lower)
 *	upperIndx[]: the upper subscript values
 *	lowerIndx[]: the lower subscript values
 *	arraytyplen: pg_type.typlen for the array type
 *	elmlen: pg_type.typlen for the array's element type
 *	elmbyval: pg_type.typbyval for the array's element type
 *	elmalign: pg_type.typalign for the array's element type
 *
 * Outputs:
 *	The return value is the new array Datum (it's never NULL)
 *
 * NOTE: we assume it is OK to scribble on the provided subscript arrays
 * lowerIndx[] and upperIndx[].  These are generally just temporaries.
 */
ArrayType* array_get_slice(ArrayType* array, int nSubscripts, int* upperIndx, int* lowerIndx, int arraytyplen,
    int elmlen, bool elmbyval, char elmalign)
{
    ArrayType* newarray = NULL;
    int i, ndim, *dim = NULL, *lb = NULL, *newlb = NULL;
    int fixedDim[1], fixedLb[1];
    Oid elemtype = InvalidOid;
    char* arraydataptr = NULL;
    bits8* arraynullsptr = NULL;
    int32 dataoffset;
    int bytes, span[MAXDIM];

    if (arraytyplen > 0) {
        /*
         * fixed-length arrays -- currently, cannot slice these because parser
         * labels output as being of the fixed-length array type! Code below
         * shows how we could support it if the parser were changed to label
         * output as a suitable varlena array type.
         */
        ereport(
            ERROR, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("slices of fixed-length arrays not implemented")));

        /*
         * fixed-length arrays -- these are assumed to be 1-d, 0-based
         *
         * XXX where would we get the correct ELEMTYPE from?
         */
        if (elmlen == 0) {
            return construct_empty_array(elemtype);
        }

        ndim = 1;
        fixedDim[0] = arraytyplen / elmlen;
        fixedLb[0] = 0;
        dim = fixedDim;
        lb = fixedLb;
        elemtype = InvalidOid; /* XXX */
        arraydataptr = (char*)array;
        arraynullsptr = NULL;
    } else {
        /* detoast input array if necessary */
        array = DatumGetArrayTypeP(PointerGetDatum(array));

        ndim = ARR_NDIM(array);
        dim = ARR_DIMS(array);
        lb = ARR_LBOUND(array);
        elemtype = ARR_ELEMTYPE(array);
        arraydataptr = ARR_DATA_PTR(array);
        arraynullsptr = ARR_NULLBITMAP(array);
    }

    /*
     * Check provided subscripts.  A slice exceeding the current array limits
     * is silently truncated to the array limits.  If we end up with an empty
     * slice, return an empty array.
     */
    if (ndim < nSubscripts || ndim <= 0 || ndim > MAXDIM)
        return construct_empty_array(elemtype);

    for (i = 0; i < nSubscripts; i++) {
        if (lowerIndx[i] < lb[i])
            lowerIndx[i] = lb[i];
        if (upperIndx[i] >= (dim[i] + lb[i]))
            upperIndx[i] = dim[i] + lb[i] - 1;
        if (lowerIndx[i] > upperIndx[i])
            return construct_empty_array(elemtype);
    }
    /* fill any missing subscript positions with full array range */
    for (; i < ndim; i++) {
        lowerIndx[i] = lb[i];
        upperIndx[i] = dim[i] + lb[i] - 1;
        if (lowerIndx[i] > upperIndx[i])
            return construct_empty_array(elemtype);
    }

    mda_get_range(ndim, span, lowerIndx, upperIndx);

    bytes =
        array_slice_size(arraydataptr, arraynullsptr, ndim, dim, lb, lowerIndx, upperIndx, elmlen, elmbyval, elmalign);

    /*
     * Currently, we put a null bitmap in the result if the source has one;
     * could be smarter ...
     */
    if (arraynullsptr != NULL) {
        dataoffset = ARR_OVERHEAD_WITHNULLS(ndim, ArrayGetNItems(ndim, span));
        bytes += dataoffset;
    } else {
        dataoffset = 0; /* marker for no null bitmap */
        bytes += ARR_OVERHEAD_NONULLS(ndim);
    }

    newarray = (ArrayType*)palloc0(bytes);
    SET_VARSIZE(newarray, bytes);
    newarray->ndim = ndim;
    newarray->dataoffset = dataoffset;
    newarray->elemtype = elemtype;
    errno_t errorno = EOK;
    errorno = memcpy_s(ARR_DIMS(newarray), ndim * sizeof(int), span, ndim * sizeof(int));
    securec_check(errorno, "\0", "\0");

    /*
     * Lower bounds of the new array are set to 1.	Formerly (before 7.3) we
     * copied the given lowerIndx values ... but that seems confusing.
     */
    newlb = ARR_LBOUND(newarray);
    for (i = 0; i < ndim; i++)
        newlb[i] = 1;

    array_extract_slice(
        newarray, ndim, dim, lb, arraydataptr, arraynullsptr, lowerIndx, upperIndx, elmlen, elmbyval, elmalign);

    return newarray;
}

/*
 * array_set :
 *		  This routine sets the value of an array element (specified by
 *		  a subscript array) to a new value specified by "dataValue".
 *
 * This handles both ordinary varlena arrays and fixed-length arrays.
 *
 * Inputs:
 *	array: the initial array object (mustn't be NULL)
 *	nSubscripts: number of subscripts supplied
 *	indx[]: the subscript values
 *	dataValue: the datum to be inserted at the given position
 *	isNull: whether dataValue is NULL
 *	arraytyplen: pg_type.typlen for the array type
 *	elmlen: pg_type.typlen for the array's element type
 *	elmbyval: pg_type.typbyval for the array's element type
 *	elmalign: pg_type.typalign for the array's element type
 *
 * Result:
 *		  A new array is returned, just like the old except for the one
 *		  modified entry.  The original array object is not changed.
 *
 * For one-dimensional arrays only, we allow the array to be extended
 * by assigning to a position outside the existing subscript range; any
 * positions between the existing elements and the new one are set to NULLs.
 *
 * NOTE: For assignments, we throw an error for invalid subscripts etc,
 * rather than returning a NULL as the fetch operations do.
 */
ArrayType* array_set(ArrayType* array, int nSubscripts, const int* indx, Datum dataValue, bool isNull, int arraytyplen,
    int elmlen, bool elmbyval, char elmalign)
{
    ArrayType* newarray = NULL;
    int i, ndim, dim[MAXDIM], lb[MAXDIM], offset;
    char* elt_ptr = NULL;
    bool newhasnulls = false;
    bits8* oldnullbitmap = NULL;
    int oldnitems, newnitems, olddatasize, newsize, olditemlen, newitemlen, overheadlen, oldoverheadlen, addedbefore,
        addedafter, lenbefore, lenafter;

    if (arraytyplen > 0) {
        /*
         * fixed-length arrays -- these are assumed to be 1-d, 0-based. We
         * cannot extend them, either.
         */
        if (nSubscripts != 1)
            ereport(ERROR, (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR), errmsg("wrong number of array subscripts")));

        if (indx[0] < 0 || elmlen == 0 || indx[0] >= arraytyplen / elmlen)
            ereport(ERROR, (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR), errmsg("array subscript out of range")));

        if (isNull)
            ereport(ERROR,
                (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
                    errmsg("cannot assign null value to an element of a fixed-length array")));

        newarray = (ArrayType*)palloc(arraytyplen);
        errno_t errorno = EOK;
        errorno = memcpy_s(newarray, arraytyplen, array, arraytyplen);
        securec_check(errorno, "\0", "\0");
        elt_ptr = (char*)newarray + indx[0] * elmlen;
        (void)ArrayCastAndSet(dataValue, elmlen, elmbyval, elmalign, elt_ptr);
        return newarray;
    }

    if (nSubscripts <= 0 || nSubscripts > MAXDIM)
        ereport(ERROR, (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR), errmsg("wrong number of array subscripts")));

    /* make sure item to be inserted is not toasted */
    if (elmlen == -1 && !isNull)
        dataValue = PointerGetDatum(PG_DETOAST_DATUM(dataValue));

    /* detoast input array if necessary */
    array = DatumGetArrayTypeP(PointerGetDatum(array));

    ndim = ARR_NDIM(array);

    /*
     * if number of dims is zero, i.e. an empty array, create an array with
     * nSubscripts dimensions, and set the lower bounds to the supplied
     * subscripts
     */
    if (ndim == 0) {
        Oid elmtype = ARR_ELEMTYPE(array);

        for (i = 0; i < nSubscripts; i++) {
            dim[i] = 1;
            lb[i] = indx[i];
        }

        return construct_md_array(&dataValue, &isNull, nSubscripts, dim, lb, elmtype, elmlen, elmbyval, elmalign);
    }

    if (ndim != nSubscripts)
        ereport(ERROR, (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR), errmsg("wrong number of array subscripts")));

    /* copy dim/lb since we may modify them */
    errno_t errorno = EOK;
    errorno = memcpy_s(dim, MAXDIM * sizeof(int), ARR_DIMS(array), ndim * sizeof(int));
    securec_check(errorno, "\0", "\0");
    errorno = memcpy_s(lb, MAXDIM * sizeof(int), ARR_LBOUND(array), ndim * sizeof(int));
    securec_check(errorno, "\0", "\0");

    newhasnulls = (ARR_HASNULL(array) || isNull);
    addedbefore = addedafter = 0;

    /*
     * Check subscripts. We assume the existing subscripts passed
     * ArrayCheckBounds, so that dim[i] + lb[i] can be computed without
     * overflow. But we must beware of other overflows in our calculations of
     * new dim[] values.
     */
    if (ndim == 1) {
        if (indx[0] < lb[0]) {
            if (pg_sub_s32_overflow(lb[0], indx[0], &addedbefore) ||
            pg_add_s32_overflow(dim[0], addedbefore, &dim[0]))
            ereport(ERROR,
                    (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                    errmsg("array size exceeds the maximum allowed (%d)",
                        (int) MaxArraySize)));
            
            lb[0] = indx[0];
            if (addedbefore > 1)
                newhasnulls = true; /* will insert nulls */
        }
        if (indx[0] >= (dim[0] + lb[0])) {
            if (pg_sub_s32_overflow(indx[0], dim[0] + lb[0], &addedafter) ||
                pg_add_s32_overflow(addedafter, 1, &addedafter) ||
                pg_add_s32_overflow(dim[0], addedafter, &dim[0]))
                ereport(ERROR,
                        (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                        errmsg("array size exceeds the maximum allowed (%d)",
                        (int) MaxArraySize)));
            if (addedafter > 1)
                newhasnulls = true; /* will insert nulls */
        }
    } else {
        /*
         * XXX currently we do not support extending multi-dimensional arrays
         * during assignment
         */
        for (i = 0; i < ndim; i++) {
            if (indx[i] < lb[i] || indx[i] >= (dim[i] + lb[i]))
                ereport(ERROR, (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR), errmsg("array subscript out of range")));
        }
    }

    /*
     * Compute sizes of items and areas to copy
     */
    newnitems = ArrayGetNItems(ndim, dim);
    ArrayCheckBounds(ndim, dim, lb);

    if (newhasnulls)
        overheadlen = ARR_OVERHEAD_WITHNULLS(ndim, newnitems);
    else
        overheadlen = ARR_OVERHEAD_NONULLS(ndim);
    oldnitems = ArrayGetNItems(ndim, ARR_DIMS(array));
    oldnullbitmap = ARR_NULLBITMAP(array);
    oldoverheadlen = ARR_DATA_OFFSET(array);
    olddatasize = ARR_SIZE(array) - oldoverheadlen;
    if (addedbefore) {
        offset = 0;
        lenbefore = 0;
        olditemlen = 0;
        lenafter = olddatasize;
    } else if (addedafter) {
        offset = oldnitems;
        lenbefore = olddatasize;
        olditemlen = 0;
        lenafter = 0;
    } else {
        offset = ArrayGetOffset(nSubscripts, dim, lb, indx);
        elt_ptr = array_seek(ARR_DATA_PTR(array), 0, oldnullbitmap, offset, elmlen, elmbyval, elmalign);
        lenbefore = (int)(elt_ptr - ARR_DATA_PTR(array));
        if (array_get_isnull(oldnullbitmap, offset))
            olditemlen = 0;
        else {
            olditemlen = att_addlength_pointer(0, elmlen, elt_ptr);
            olditemlen = att_align_nominal(olditemlen, elmalign);
        }
        lenafter = (int)(olddatasize - lenbefore - olditemlen);
    }

    if (isNull)
        newitemlen = 0;
    else {
        newitemlen = att_addlength_datum(0, elmlen, dataValue);
        newitemlen = att_align_nominal(newitemlen, elmalign);
    }

    newsize = overheadlen + lenbefore + newitemlen + lenafter;

    /*
     * OK, create the new array and fill in header/dimensions
     */
    newarray = (ArrayType*)palloc0(newsize);
    SET_VARSIZE(newarray, newsize);
    newarray->ndim = ndim;
    newarray->dataoffset = newhasnulls ? overheadlen : 0;
    newarray->elemtype = ARR_ELEMTYPE(array);
    errorno = memcpy_s(ARR_DIMS(newarray), ndim * sizeof(int), dim, ndim * sizeof(int));
    securec_check(errorno, "\0", "\0");
    errorno = memcpy_s(ARR_LBOUND(newarray), ndim * sizeof(int), lb, ndim * sizeof(int));
    securec_check(errorno, "\0", "\0");

    /*
     * Fill in data
     */
    if (lenbefore > 0) {
        errorno = memcpy_s((char*)newarray + overheadlen, lenbefore, (char*)array + oldoverheadlen, lenbefore);
        securec_check(errorno, "\0", "\0");
    }
    if (!isNull)
        (void)ArrayCastAndSet(dataValue, elmlen, elmbyval, elmalign, (char*)newarray + overheadlen + lenbefore);
    if (lenafter > 0) {
        errorno = memcpy_s((char*)newarray + overheadlen + lenbefore + newitemlen,
            lenafter,
            (char*)array + oldoverheadlen + lenbefore + olditemlen,
            lenafter);
        securec_check(errorno, "\0", "\0");
    }

    /*
     * Fill in nulls bitmap if needed
     *
     * Note: it's possible we just replaced the last NULL with a non-NULL, and
     * could get rid of the bitmap.  Seems not worth testing for though.
     */
    if (newhasnulls) {
        bits8* newnullbitmap = ARR_NULLBITMAP(newarray);

        if (newnullbitmap != NULL) {
            /* Zero the bitmap to take care of marking inserted positions null */
            errno_t errorno = EOK;
            errorno = memset_s(newnullbitmap, (newnitems + 7) / 8, 0, (newnitems + 7) / 8);
            securec_check(errorno, "\0", "\0");
        }
        /* Fix the inserted value */
        if (addedafter)
            array_set_isnull(newnullbitmap, newnitems - 1, isNull);
        else
            array_set_isnull(newnullbitmap, offset, isNull);
        /* Fix the copied range(s) */
        if (addedbefore)
            array_bitmap_copy(newnullbitmap, addedbefore, oldnullbitmap, 0, oldnitems);
        else {
            array_bitmap_copy(newnullbitmap, 0, oldnullbitmap, 0, offset);
            if (addedafter == 0)
                array_bitmap_copy(newnullbitmap, offset + 1, oldnullbitmap, offset + 1, oldnitems - offset - 1);
        }
    }

    return newarray;
}

/*
 * array_set_slice :
 *		  This routine sets the value of a range of array locations (specified
 *		  by upper and lower subscript values) to new values passed as
 *		  another array.
 *
 * This handles both ordinary varlena arrays and fixed-length arrays.
 *
 * Inputs:
 *	array: the initial array object (mustn't be NULL)
 *	nSubscripts: number of subscripts supplied (must be same for upper/lower)
 *	upperIndx[]: the upper subscript values
 *	lowerIndx[]: the lower subscript values
 *	srcArray: the source for the inserted values
 *	isNull: indicates whether srcArray is NULL
 *	arraytyplen: pg_type.typlen for the array type
 *	elmlen: pg_type.typlen for the array's element type
 *	elmbyval: pg_type.typbyval for the array's element type
 *	elmalign: pg_type.typalign for the array's element type
 *
 * Result:
 *		  A new array is returned, just like the old except for the
 *		  modified range.  The original array object is not changed.
 *
 * For one-dimensional arrays only, we allow the array to be extended
 * by assigning to positions outside the existing subscript range; any
 * positions between the existing elements and the new ones are set to NULLs.
 *
 * NOTE: we assume it is OK to scribble on the provided index arrays
 * lowerIndx[] and upperIndx[].  These are generally just temporaries.
 *
 * NOTE: For assignments, we throw an error for silly subscripts etc,
 * rather than returning a NULL or empty array as the fetch operations do.
 */
ArrayType* array_set_slice(ArrayType* array, int nSubscripts, int* upperIndx, int* lowerIndx, ArrayType* srcArray,
    bool isNull, int arraytyplen, int elmlen, bool elmbyval, char elmalign)
{
    ArrayType* newarray = NULL;
    int i, ndim, dim[MAXDIM], lb[MAXDIM], span[MAXDIM];
    bool newhasnulls = false;
    int nitems, nsrcitems, olddatasize, newsize, olditemsize, newitemsize, overheadlen, oldoverheadlen, addedbefore,
        addedafter, lenbefore, lenafter, itemsbefore, itemsafter, nolditems;

    /* Currently, assignment from a NULL source array is a no-op */
    if (isNull)
        return array;

    if (arraytyplen > 0) {
        /*
         * fixed-length arrays -- not got round to doing this...
         */
        ereport(ERROR,
            (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                errmsg("updates on slices of fixed-length arrays not implemented")));
    }

    /* detoast arrays if necessary */
    array = DatumGetArrayTypeP(PointerGetDatum(array));
    srcArray = DatumGetArrayTypeP(PointerGetDatum(srcArray));

    /* note: we assume srcArray contains no toasted elements */

    ndim = ARR_NDIM(array);

    /*
     * if number of dims is zero, i.e. an empty array, create an array with
     * nSubscripts dimensions, and set the upper and lower bounds to the
     * supplied subscripts
     */
    if (ndim == 0) {
        Datum* dvalues = NULL;
        bool* dnulls = NULL;
        int nelems;
        Oid elmtype = ARR_ELEMTYPE(array);

        deconstruct_array(srcArray, elmtype, elmlen, elmbyval, elmalign, &dvalues, &dnulls, &nelems);

        for (i = 0; i < nSubscripts; i++) {
            dim[i] = 1 + upperIndx[i] - lowerIndx[i];
            lb[i] = lowerIndx[i];
        }

        /* complain if too few source items; we ignore extras, however */
        if (nelems < ArrayGetNItems(nSubscripts, dim))
            ereport(ERROR, (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR), errmsg("source array too small")));

        return construct_md_array(dvalues, dnulls, nSubscripts, dim, lb, elmtype, elmlen, elmbyval, elmalign);
    }

    if (ndim < nSubscripts || ndim <= 0 || ndim > MAXDIM)
        ereport(ERROR, (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR), errmsg("wrong number of array subscripts")));

    /* copy dim/lb since we may modify them */
    errno_t errorno = EOK;
    errorno = memcpy_s(dim, ndim * sizeof(int), ARR_DIMS(array), ndim * sizeof(int));
    securec_check(errorno, "\0", "\0");
    errorno = memcpy_s(lb, ndim * sizeof(int), ARR_LBOUND(array), ndim * sizeof(int));
    securec_check(errorno, "\0", "\0");

    newhasnulls = (ARR_HASNULL(array) || ARR_HASNULL(srcArray));
    addedbefore = addedafter = 0;

    /*
     * Check subscripts. We assume the existing subscripts passed
     * ArrayCheckBounds, so that dim[i] + lb[i] can be computed without
     * overflow. But we must beware of other overflows in our calculations of
     * new dim[] values.
     */
    if (ndim == 1) {
        Assert(nSubscripts == 1);
        if (lowerIndx[0] > upperIndx[0])
            ereport(
                ERROR, (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR), errmsg("upper bound cannot be less than lower bound")));
        if (lowerIndx[0] < lb[0]) {
            if (pg_sub_s32_overflow(lb[0], lowerIndx[0], &addedbefore) ||
                pg_add_s32_overflow(dim[0], addedbefore, &dim[0]))
                ereport(ERROR,
                        (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                        errmsg("array size exceeds the maximum allowed (%d)",
                               (int) MaxArraySize)));
            lb[0] = lowerIndx[0];
            if (addedbefore > 1)
                newhasnulls = true; /*will insert nulls*/
        }
        if (upperIndx[0] >= (dim[0] + lb[0])) {
            if (pg_sub_s32_overflow(upperIndx[0], dim[0] + lb[0], &addedafter) ||
                pg_add_s32_overflow(addedafter, 1, &addedafter) ||
                pg_add_s32_overflow(dim[0], addedafter, &dim[0]))
                ereport(ERROR,
                        (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                        errmsg("array size exceeds the maximum allowed (%d)",
                        (int) MaxArraySize)));
            if (addedafter > 1)
                newhasnulls = true; /* will insert nulls */
        }
    } else {
        /*
         * XXX currently we do not support extending multi-dimensional arrays
         * during assignment
         */
        for (i = 0; i < nSubscripts; i++) {
            if (lowerIndx[i] > upperIndx[i])
                ereport(ERROR,
                    (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR), errmsg("upper bound cannot be less than lower bound")));
            if (lowerIndx[i] < lb[i] || upperIndx[i] >= (dim[i] + lb[i]))
                ereport(ERROR, (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR), errmsg("array subscript out of range")));
        }
        /* fill any missing subscript positions with full array range */
        for (; i < ndim; i++) {
            lowerIndx[i] = lb[i];
            upperIndx[i] = dim[i] + lb[i] - 1;
            if (lowerIndx[i] > upperIndx[i])
                ereport(ERROR,
                    (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR), errmsg("upper bound cannot be less than lower bound")));
        }
    }

    /* Do this mainly to check for overflow */
    nitems = ArrayGetNItems(ndim, dim);
    ArrayCheckBounds(ndim, dim, lb);

    /*
     * Make sure source array has enough entries.  Note we ignore the shape of
     * the source array and just read entries serially.
     */
    mda_get_range(ndim, span, lowerIndx, upperIndx);
    nsrcitems = ArrayGetNItems(ndim, span);
    if (nsrcitems > ArrayGetNItems(ARR_NDIM(srcArray), ARR_DIMS(srcArray)))
        ereport(ERROR, (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR), errmsg("source array too small")));

    /*
     * Compute space occupied by new entries, space occupied by replaced
     * entries, and required space for new array.
     */
    if (newhasnulls)
        overheadlen = ARR_OVERHEAD_WITHNULLS(ndim, nitems);
    else
        overheadlen = ARR_OVERHEAD_NONULLS(ndim);
    newitemsize =
        array_nelems_size(ARR_DATA_PTR(srcArray), 0, ARR_NULLBITMAP(srcArray), nsrcitems, elmlen, elmbyval, elmalign);
    oldoverheadlen = ARR_DATA_OFFSET(array);
    olddatasize = ARR_SIZE(array) - oldoverheadlen;
    if (ndim > 1) {
        /*
         * here we do not need to cope with extension of the array; it would
         * be a lot more complicated if we had to do so...
         */
        olditemsize = array_slice_size(ARR_DATA_PTR(array),
            ARR_NULLBITMAP(array),
            ndim,
            dim,
            lb,
            lowerIndx,
            upperIndx,
            elmlen,
            elmbyval,
            elmalign);
        lenbefore = lenafter = 0; /* keep compiler quiet */
        itemsbefore = itemsafter = nolditems = 0;
    } else {
        /*
         * here we must allow for possibility of slice larger than orig array
         * and/or not adjacent to orig array subscripts
         */
        int oldlb = ARR_LBOUND(array)[0];
        int oldub = oldlb + ARR_DIMS(array)[0] - 1;
        int slicelb = Max(oldlb, lowerIndx[0]);
        int sliceub = Min(oldub, upperIndx[0]);
        char* oldarraydata = ARR_DATA_PTR(array);
        bits8* oldarraybitmap = ARR_NULLBITMAP(array);

        /* count/size of old array entries that will go before the slice */
        itemsbefore = Min(slicelb, oldub + 1) - oldlb;
        lenbefore = array_nelems_size(oldarraydata, 0, oldarraybitmap, itemsbefore, elmlen, elmbyval, elmalign);
        /* count/size of old array entries that will be replaced by slice */
        if (slicelb > sliceub) {
            nolditems = 0;
            olditemsize = 0;
        } else {
            nolditems = sliceub - slicelb + 1;
            olditemsize = array_nelems_size(
                oldarraydata + lenbefore, itemsbefore, oldarraybitmap, nolditems, elmlen, elmbyval, elmalign);
        }
        /* count/size of old array entries that will go after the slice */
        itemsafter = oldub + 1 - Max(sliceub + 1, oldlb);
        lenafter = olddatasize - lenbefore - olditemsize;
    }

    newsize = overheadlen + olddatasize - olditemsize + newitemsize;

    newarray = (ArrayType*)palloc0(newsize);
    SET_VARSIZE(newarray, newsize);
    newarray->ndim = ndim;
    newarray->dataoffset = newhasnulls ? overheadlen : 0;
    newarray->elemtype = ARR_ELEMTYPE(array);
    errorno = memcpy_s(ARR_DIMS(newarray), ndim * sizeof(int), dim, ndim * sizeof(int));
    securec_check(errorno, "\0", "\0");
    errorno = memcpy_s(ARR_LBOUND(newarray), ndim * sizeof(int), lb, ndim * sizeof(int));
    securec_check(errorno, "\0", "\0");

    if (ndim > 1) {
        /*
         * here we do not need to cope with extension of the array; it would
         * be a lot more complicated if we had to do so...
         */
        array_insert_slice(newarray, array, srcArray, ndim, dim, lb, lowerIndx, upperIndx, elmlen, elmbyval, elmalign);
    } else {
        /* fill in data */
        errno_t errorno = EOK;
        if (lenbefore > 0) {
            errorno = memcpy_s((char*)newarray + overheadlen, lenbefore, (char*)array + oldoverheadlen, lenbefore);
            securec_check(errorno, "\0", "\0");
        }
        if (newitemsize > 0) {
            errorno =
                memcpy_s((char*)newarray + overheadlen + lenbefore, newitemsize, ARR_DATA_PTR(srcArray), newitemsize);
            securec_check(errorno, "\0", "\0");
        }
        if (lenafter > 0) {
            errorno = memcpy_s((char*)newarray + overheadlen + lenbefore + newitemsize,
                lenafter,
                (char*)array + oldoverheadlen + lenbefore + olditemsize,
                lenafter);
            securec_check(errorno, "\0", "\0");
        }
        /* fill in nulls bitmap if needed */
        if (newhasnulls) {
            bits8* newnullbitmap = ARR_NULLBITMAP(newarray);
            bits8* oldnullbitmap = ARR_NULLBITMAP(array);

            if (NULL == newnullbitmap)
                ereport(ERROR, (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED), errmsg("null pointer for nullbitmap.")));
            /* Zero the bitmap to handle marking inserted positions null */
            errno_t errorno = EOK;
            errorno = memset_s(newnullbitmap, (nitems + 7) / 8, 0, (nitems + 7) / 8);
            securec_check(errorno, "\0", "\0");
            array_bitmap_copy(newnullbitmap, addedbefore, oldnullbitmap, 0, itemsbefore);
            array_bitmap_copy(newnullbitmap, lowerIndx[0] - lb[0], ARR_NULLBITMAP(srcArray), 0, nsrcitems);
            array_bitmap_copy(newnullbitmap,
                addedbefore + itemsbefore + nolditems,
                oldnullbitmap,
                itemsbefore + nolditems,
                itemsafter);
        }
    }

    return newarray;
}

/*
 * array_map()
 *
 * Map an array through an arbitrary function.	Return a new array with
 * same dimensions and each source element transformed by fn().  Each
 * source element is passed as the first argument to fn(); additional
 * arguments to be passed to fn() can be specified by the caller.
 * The output array can have a different element type than the input.
 *
 * Parameters are:
 * * fcinfo: a function-call data structure pre-constructed by the caller
 *	 to be ready to call the desired function, with everything except the
 *	 first argument position filled in.  In particular, flinfo identifies
 *	 the function fn(), and if nargs > 1 then argument positions after the
 *	 first must be preset to the additional values to be passed.  The
 *	 first argument position initially holds the input array value.
 * * inpType: OID of element type of input array.  This must be the same as,
 *	 or binary-compatible with, the first argument type of fn().
 * * retType: OID of element type of output array.	This must be the same as,
 *	 or binary-compatible with, the result type of fn().
 * * amstate: workspace for array_map.	Must be zeroed by caller before
 *	 first call, and not touched after that.
 *
 * It is legitimate to pass a freshly-zeroed ArrayMapState on each call,
 * but better performance can be had if the state can be preserved across
 * a series of calls.
 *
 * NB: caller must assure that input array is not NULL.  NULL elements in
 * the array are OK however.
 */
Datum array_map(FunctionCallInfo fcinfo, Oid inpType, Oid retType, ArrayMapState* amstate)
{
    ArrayType* v = NULL;
    ArrayType* result = NULL;
    Datum* values = NULL;
    bool* nulls = NULL;
    Datum elt;
    int* dim = NULL;
    int ndim;
    int nitems;
    int i;
    int32 nbytes = 0;
    int32 dataoffset;
    bool hasnulls = false;
    int inp_typlen;
    bool inp_typbyval = false;
    char inp_typalign;
    int typlen;
    bool typbyval = false;
    char typalign;
    char* s = NULL;
    bits8* bitmap = NULL;
    uint32 bitmask;
    ArrayMetaState* inp_extra = NULL;
    ArrayMetaState* ret_extra = NULL;

    /* Get input array */
    if (fcinfo->nargs < 1)
        ereport(ERROR, (errcode(ERRCODE_DATA_EXCEPTION), errmsg("invalid nargs: %d", fcinfo->nargs)));

    if (PG_ARGISNULL(0))
        ereport(ERROR, (errcode(ERRCODE_ARRAY_ELEMENT_ERROR), errmsg("null input array")));

    v = PG_GETARG_ARRAYTYPE_P(0);

    Assert(ARR_ELEMTYPE(v) == inpType);

    ndim = ARR_NDIM(v);
    dim = ARR_DIMS(v);
    nitems = ArrayGetNItems(ndim, dim);

    /* Check for empty array */
    if (nitems <= 0) {
        /* Return empty array */
        PG_RETURN_ARRAYTYPE_P(construct_empty_array(retType));
    }

    /*
     * We arrange to look up info about input and return element types only
     * once per series of calls, assuming the element type doesn't change
     * underneath us.
     */
    inp_extra = &amstate->inp_extra;
    ret_extra = &amstate->ret_extra;

    if (inp_extra->element_type != inpType) {
        get_typlenbyvalalign(inpType, &inp_extra->typlen, &inp_extra->typbyval, &inp_extra->typalign);
        inp_extra->element_type = inpType;
    }
    inp_typlen = inp_extra->typlen;
    inp_typbyval = inp_extra->typbyval;
    inp_typalign = inp_extra->typalign;

    if (ret_extra->element_type != retType) {
        get_typlenbyvalalign(retType, &ret_extra->typlen, &ret_extra->typbyval, &ret_extra->typalign);
        ret_extra->element_type = retType;
    }
    typlen = ret_extra->typlen;
    typbyval = ret_extra->typbyval;
    typalign = ret_extra->typalign;

    /* Allocate temporary arrays for new values */
    values = (Datum*)palloc(nitems * sizeof(Datum));
    nulls = (bool*)palloc(nitems * sizeof(bool));

    /* Loop over source data */
    s = ARR_DATA_PTR(v);
    bitmap = ARR_NULLBITMAP(v);
    bitmask = 1;
    hasnulls = false;

    for (i = 0; i < nitems; i++) {
        bool callit = true;

        /* Get source element, checking for NULL */
        if (bitmap && (*bitmap & bitmask) == 0) {
            fcinfo->argnull[0] = true;
        } else {
            elt = fetch_att(s, inp_typbyval, inp_typlen);
            s = att_addlength_datum(s, inp_typlen, elt);
            s = (char*)att_align_nominal(s, inp_typalign);
            fcinfo->arg[0] = elt;
            fcinfo->argnull[0] = false;
        }

        /*
         * Apply the given function to source elt and extra args.
         */
        if (fcinfo->flinfo->fn_strict) {
            int j;

            for (j = 0; j < fcinfo->nargs; j++) {
                if (fcinfo->argnull[j]) {
                    callit = false;
                    break;
                }
            }
        }

        if (callit) {
            fcinfo->isnull = false;
            /* The input parameters are transferred by ExecEvalArrayCoerceExpr.
             * The number of input parameters is fixed to three.
             * However, the number of input parameters required by the function is not necessarily three.
             * Therefore, the number of input parameters required by the function is assigned to fcinfo->nargs.
             */
            int oldnargs = fcinfo->nargs;
            fcinfo->nargs = fcinfo->flinfo->fn_nargs;
            values[i] = FunctionCallInvoke(fcinfo);
            fcinfo->nargs = oldnargs;
        } else
            fcinfo->isnull = true;

        nulls[i] = fcinfo->isnull;
        if (fcinfo->isnull)
            hasnulls = true;
        else {
            /* Ensure data is not toasted */
            if (typlen == -1)
                values[i] = PointerGetDatum(PG_DETOAST_DATUM(values[i]));
            /* Update total result size */
            nbytes = att_addlength_datum(nbytes, typlen, values[i]);
            nbytes = att_align_nominal(nbytes, typalign);
            /* check for overflow of total request */
            if (!AllocSizeIsValid(nbytes))
                ereport(ERROR,
                    (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                        errmsg("array size exceeds the maximum allowed (%d)", (int)MaxAllocSize)));
        }

        /* advance bitmap pointer if any */
        if (bitmap != NULL) {
            bitmask <<= 1;
            if (bitmask == 0x100) {
                bitmap++;
                bitmask = 1;
            }
        }
    }

    /* Allocate and initialize the result array */
    if (hasnulls) {
        dataoffset = ARR_OVERHEAD_WITHNULLS(ndim, nitems);
        nbytes += dataoffset;
    } else {
        dataoffset = 0; /* marker for no null bitmap */
        nbytes += ARR_OVERHEAD_NONULLS(ndim);
    }
    result = (ArrayType*)palloc0(nbytes);
    SET_VARSIZE(result, nbytes);
    result->ndim = ndim;
    result->dataoffset = dataoffset;
    result->elemtype = retType;
    errno_t errorno = EOK;
    errorno = memcpy_s(ARR_DIMS(result), 2 * ndim * sizeof(int), ARR_DIMS(v), 2 * ndim * sizeof(int));
    securec_check(errorno, "\0", "\0");

    /*
     * Note: do not risk trying to pfree the results of the called function
     */
    CopyArrayEls(result, values, nulls, nitems, typlen, typbyval, typalign, false);

    pfree(values);
    pfree(nulls);

    PG_RETURN_ARRAYTYPE_P(result);
}

/*
 * construct_array	--- simple method for constructing an array object
 *
 * elems: array of Datum items to become the array contents
 *		  (NULL element values are not supported).
 * nelems: number of items
 * elmtype, elmlen, elmbyval, elmalign: info for the datatype of the items
 *
 * A palloc'd 1-D array object is constructed and returned.  Note that
 * elem values will be copied into the object even if pass-by-ref type.
 *
 * NOTE: it would be cleaner to look up the elmlen/elmbval/elmalign info
 * from the system catalogs, given the elmtype.  However, the caller is
 * in a better position to cache this info across multiple uses, or even
 * to hard-wire values if the element type is hard-wired.
 */
ArrayType* construct_array(Datum* elems, int nelems, Oid elmtype, int elmlen, bool elmbyval, char elmalign)
{
    int dims[1];
    int lbs[1];

    dims[0] = nelems;
    lbs[0] = 1;

    return construct_md_array(elems, NULL, 1, dims, lbs, elmtype, elmlen, elmbyval, elmalign);
}

/*
 * construct_md_array	--- simple method for constructing an array object
 *							with arbitrary dimensions and possible NULLs
 *
 * elems: array of Datum items to become the array contents
 * nulls: array of is-null flags (can be NULL if no nulls)
 * ndims: number of dimensions
 * dims: integer array with size of each dimension
 * lbs: integer array with lower bound of each dimension
 * elmtype, elmlen, elmbyval, elmalign: info for the datatype of the items
 *
 * A palloc'd ndims-D array object is constructed and returned.  Note that
 * elem values will be copied into the object even if pass-by-ref type.
 *
 * NOTE: it would be cleaner to look up the elmlen/elmbval/elmalign info
 * from the system catalogs, given the elmtype.  However, the caller is
 * in a better position to cache this info across multiple uses, or even
 * to hard-wire values if the element type is hard-wired.
 */
ArrayType* construct_md_array(Datum* elems, bool* nulls, int ndims, int* dims, const int* lbs, Oid elmtype, int elmlen,
    bool elmbyval, char elmalign)
{
    ArrayType* result = NULL;
    bool hasnulls = false;
    int32 nbytes;
    int32 dataoffset;
    int i;
    int nelems;

    if (ndims < 0) /* we do allow zero-dimension arrays */
        ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("invalid number of dimensions: %d", ndims)));
    if (ndims > MAXDIM)
        ereport(ERROR,
            (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                errmsg("number of array dimensions (%d) exceeds the maximum allowed (%d)", ndims, MAXDIM)));
    if (elems == NULL)
        ereport(ERROR,
            (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                errmsg(
                    "the object array should be consist of some Datum items, otherwise use construct_empty_array.")));

    /* fast track for empty array */
    if (ndims == 0)
        return construct_empty_array(elmtype);

    nelems = ArrayGetNItems(ndims, dims);
    ArrayCheckBounds(ndims, dims, lbs);
    
    /* compute required space */
    nbytes = 0;
    hasnulls = false;
    for (i = 0; i < nelems; i++) {
        if (nulls != NULL && nulls[i]) {
            hasnulls = true;
            continue;
        }
        /* make sure data is not toasted */
        if (elmlen == -1)
            elems[i] = PointerGetDatum(PG_DETOAST_DATUM(elems[i]));
        nbytes = att_addlength_datum(nbytes, elmlen, elems[i]);
        nbytes = att_align_nominal(nbytes, elmalign);
        /* check for overflow of total request */
        if (!AllocSizeIsValid(nbytes))
            ereport(ERROR,
                (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                    errmsg("array size exceeds the maximum allowed (%d)", (int)MaxAllocSize)));
    }

    /* Allocate and initialize result array */
    if (hasnulls) {
        dataoffset = ARR_OVERHEAD_WITHNULLS(ndims, nelems);
        nbytes += dataoffset;
    } else {
        dataoffset = 0; /* marker for no null bitmap */
        nbytes += ARR_OVERHEAD_NONULLS(ndims);
    }
    result = (ArrayType*)palloc0(nbytes);
    SET_VARSIZE(result, nbytes);
    result->ndim = ndims;
    result->dataoffset = dataoffset;
    result->elemtype = elmtype;
    errno_t errorno = EOK;
    errorno = memcpy_s(ARR_DIMS(result), ndims * sizeof(int), dims, ndims * sizeof(int));
    securec_check(errorno, "\0", "\0");
    errorno = memcpy_s(ARR_LBOUND(result), ndims * sizeof(int), lbs, ndims * sizeof(int));
    securec_check(errorno, "\0", "\0");

    CopyArrayEls(result, elems, nulls, nelems, elmlen, elmbyval, elmalign, false);

    return result;
}

/*
 * construct_empty_array	--- make a zero-dimensional array of given type
 */
ArrayType* construct_empty_array(Oid elmtype)
{
    ArrayType* result = NULL;

    result = (ArrayType*)palloc0(sizeof(ArrayType));
    SET_VARSIZE(result, sizeof(ArrayType));
    result->ndim = 0;
    result->dataoffset = 0;
    result->elemtype = elmtype;
    return result;
}

/*
 * deconstruct_array  --- simple method for extracting data from an array
 *
 * array: array object to examine (must not be NULL)
 * elmtype, elmlen, elmbyval, elmalign: info for the datatype of the items
 * elemsp: return value, set to point to palloc'd array of Datum values
 * nullsp: return value, set to point to palloc'd array of isnull markers
 * nelemsp: return value, set to number of extracted values
 *
 * The caller may pass nullsp == NULL if it does not support NULLs in the
 * array.  Note that this produces a very uninformative error message,
 * so do it only in cases where a NULL is really not expected.
 *
 * If array elements are pass-by-ref data type, the returned Datums will
 * be pointers into the array object.
 *
 * NOTE: it would be cleaner to look up the elmlen/elmbval/elmalign info
 * from the system catalogs, given the elmtype.  However, in most current
 * uses the type is hard-wired into the caller and so we can save a lookup
 * cycle by hard-wiring the type info as well.
 */
void deconstruct_array(ArrayType* array, Oid elmtype, int elmlen, bool elmbyval, char elmalign, Datum** elemsp,
    bool** nullsp, int* nelemsp)
{
    Datum* elems = NULL;
    bool* nulls = NULL;
    int nelems;
    char* p = NULL;
    bits8* bitmap = NULL;
    uint32 bitmask;
    int i;

    Assert(ARR_ELEMTYPE(array) == elmtype);

    nelems = ArrayGetNItems(ARR_NDIM(array), ARR_DIMS(array));
    *elemsp = elems = (Datum*)palloc(nelems * sizeof(Datum));
    if (nullsp != NULL)
        *nullsp = nulls = (bool*)palloc0(nelems * sizeof(bool));
    else
        nulls = NULL;
    *nelemsp = nelems;

    p = ARR_DATA_PTR(array);
    bitmap = ARR_NULLBITMAP(array);
    bitmask = 1;

    for (i = 0; i < nelems; i++) {
        /* Get source element, checking for NULL */
        if (bitmap && (*bitmap & bitmask) == 0) {
            elems[i] = (Datum)0;
            if (nulls != NULL)
                nulls[i] = true;
            else
                ereport(ERROR,
                    (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED),
                        errmsg("null array element not allowed in this context")));
        } else {
            elems[i] = fetch_att(p, elmbyval, elmlen);
            p = att_addlength_pointer(p, elmlen, p);
            p = (char*)att_align_nominal(p, elmalign);
        }

        /* advance bitmap pointer if any */
        if (bitmap != NULL) {
            bitmask <<= 1;
            if (bitmask == 0x100) {
                bitmap++;
                bitmask = 1;
            }
        }
    }
}

/*
 * array_contains_nulls --- detect whether an array has any null elements
 *
 * This gives an accurate answer, whereas testing ARR_HASNULL only tells
 * if the array *might* contain a null.
 */
bool array_contains_nulls(ArrayType* array)
{
    int nelems;
    bits8* bitmap = NULL;
    uint32 bitmask;

    /* Easy answer if there's no null bitmap */
    if (array == NULL || !ARR_HASNULL(array))
        return false;

    nelems = ArrayGetNItems(ARR_NDIM(array), ARR_DIMS(array));

    bitmap = ARR_NULLBITMAP(array);

    /* check whole bytes of the bitmap byte-at-a-time */
    while (nelems >= 8) {
        if (*bitmap != 0xFF)
            return true;
        bitmap++;
        nelems -= 8;
    }

    /* check last partial byte */
    bitmask = 1;
    while (nelems > 0) {
        if ((*bitmap & bitmask) == 0)
            return true;
        bitmask <<= 1;
        nelems--;
    }

    return false;
}

static bool array_eq_inner(ArrayType* array1, ArrayType* array2, Oid collation, FunctionCallInfoData* fcinfo)
{
    int ndims1 = ARR_NDIM(array1);
    int ndims2 = ARR_NDIM(array2);
    int* dims1 = ARR_DIMS(array1);
    int* dims2 = ARR_DIMS(array2);
    Oid element_type = ARR_ELEMTYPE(array1);
    bool result = true;
    int nitems;
    TypeCacheEntry* typentry = NULL;
    int typlen;
    bool typbyval = false;
    char typalign;
    char* ptr1 = NULL;
    char* ptr2 = NULL;
    bits8* bitmap1 = NULL;
    bits8* bitmap2 = NULL;
    uint32 bitmask;
    int i;
    FunctionCallInfoData locfcinfo;

    if (element_type != ARR_ELEMTYPE(array2))
        ereport(
            ERROR, (errcode(ERRCODE_DATATYPE_MISMATCH), errmsg("cannot compare arrays of different element types")));

    /* fast path if the arrays do not have the same dimensionality */
    if (ndims1 != ndims2 || memcmp(dims1, dims2, 2 * ndims1 * sizeof(int)) != 0) {
        result = false;
    } else {
        /*
         * We arrange to look up the equality function only once per series of
         * calls, assuming the element type doesn't change underneath us.  The
         * typcache is used so that we have no memory leakage when being used
         * as an index support function.
         */
        typentry = (TypeCacheEntry*)fcinfo->flinfo->fn_extra;
        if (typentry == NULL || typentry->type_id != element_type) {
            typentry = lookup_type_cache(element_type, TYPECACHE_EQ_OPR_FINFO);
            if (!OidIsValid(typentry->eq_opr_finfo.fn_oid))
                ereport(ERROR,
                    (errcode(ERRCODE_UNDEFINED_FUNCTION),
                        errmsg("could not identify an equality operator for type %s", format_type_be(element_type))));
            fcinfo->flinfo->fn_extra = (void*)typentry;
        }
        typlen = typentry->typlen;
        typbyval = typentry->typbyval;
        typalign = typentry->typalign;

        /*
         * apply the operator to each pair of array elements.
         */
        InitFunctionCallInfoData(locfcinfo, &typentry->eq_opr_finfo, TWO_ARGS, collation, NULL, NULL);

        /* Loop over source data */
        nitems = ArrayGetNItems(ndims1, dims1);
        ptr1 = ARR_DATA_PTR(array1);
        ptr2 = ARR_DATA_PTR(array2);
        bitmap1 = ARR_NULLBITMAP(array1);
        bitmap2 = ARR_NULLBITMAP(array2);
        bitmask = 1; /* use same bitmask for both arrays */

        for (i = 0; i < nitems; i++) {
            Datum elt1;
            Datum elt2;
            bool isnull1 = false;
            bool isnull2 = false;
            bool oprresult = false;

            /* Get elements, checking for NULL */
            if (bitmap1 && (*bitmap1 & bitmask) == 0) {
                isnull1 = true;
                elt1 = (Datum)0;
            } else {
                isnull1 = false;
                elt1 = fetch_att(ptr1, typbyval, typlen);
                ptr1 = att_addlength_pointer(ptr1, typlen, ptr1);
                ptr1 = (char*)att_align_nominal(ptr1, typalign);
            }

            if (bitmap2 && (*bitmap2 & bitmask) == 0) {
                isnull2 = true;
                elt2 = (Datum)0;
            } else {
                isnull2 = false;
                elt2 = fetch_att(ptr2, typbyval, typlen);
                ptr2 = att_addlength_pointer(ptr2, typlen, ptr2);
                ptr2 = (char*)att_align_nominal(ptr2, typalign);
            }

            /* advance bitmap pointers if any */
            bitmask <<= 1;
            if (bitmask == 0x100) {
                if (bitmap1 != NULL) {
                    bitmap1++;
                }
                if (bitmap2 != NULL) {
                    bitmap2++;
                }
                bitmask = 1;
            }

            /*
             * We consider two NULLs equal; NULL and not-NULL are unequal.
             */
            if (isnull1 && isnull2) {
                continue;
            }
            if (isnull1 || isnull2) {
                result = false;
                break;
            }

            /*
             * Apply the operator to the element pair
             */
            locfcinfo.arg[0] = elt1;
            locfcinfo.arg[1] = elt2;
            locfcinfo.argnull[0] = false;
            locfcinfo.argnull[1] = false;
            locfcinfo.isnull = false;
            oprresult = DatumGetBool(FunctionCallInvoke(&locfcinfo));
            if (!oprresult) {
                result = false;
                break;
            }
        }
    }

    /* Avoid leaking memory when handed toasted input. */
    PG_FREE_IF_COPY(array1, 0);
    PG_FREE_IF_COPY(array2, 1);
    return result;
}

static int CmpArrayItem(const void* a, const void* b, void* arg)
{
    CmpFuncArgs* cmpFuncArgs = (CmpFuncArgs*)arg;
    FunctionCallInfoData locfcinfo = cmpFuncArgs->locfcinfo;
    FmgrInfo func = cmpFuncArgs->func;
    Oid collation = cmpFuncArgs->collation;
    ArrayItem* left = (ArrayItem*)a;
    ArrayItem* right = (ArrayItem*)b;

    InitFunctionCallInfoData(locfcinfo, &func, TWO_ARGS, collation, NULL, NULL);

    locfcinfo.arg[0] = left->elt;
    locfcinfo.arg[1] = right->elt;
    locfcinfo.argnull[0] = false;
    locfcinfo.argnull[1] = false;
    locfcinfo.isnull = false;
    return DatumGetInt32(FunctionCallInvoke(&locfcinfo));
}

static void inner_init_element(bool isnull, Datum* elt, char** ptr, ArrayItem* arr, 
                               int typlen, bool typbyval, char typalign)
{
    if (isnull) {
        *elt = (Datum)0;
        arr->isnull = true;
    } else {
        *elt = fetch_att(*ptr, typbyval, typlen);
        arr->isnull = false;
        *ptr = att_addlength_pointer(*ptr, typlen, *ptr);
        *ptr = (char*)att_align_nominal(*ptr, typalign);
    }
}

/*
 * When in compatible_od_db_array mode, use this function to compare two arrays. 
 * If the elements in the array are the same, then regardless of their order, 
 * they are equal
 */
static bool array_eq_db_a_inner(ArrayType* array1, ArrayType* array2, Oid collation, FunctionCallInfoData* fcinfo)
{
    int ndims1 = ARR_NDIM(array1);
    int ndims2 = ARR_NDIM(array2);
    int* dims1 = ARR_DIMS(array1);
    int* dims2 = ARR_DIMS(array2);
    Oid element_type = ARR_ELEMTYPE(array1);
    bool result = true;
    TypeCacheEntry* typentry = NULL;
    int typlen = 0;
    bool typbyval = false;
    char typalign = 0;
    TypeCacheEntry* typentry_sort = NULL;
    HeapTuple opertup;
    FunctionCallInfoData locfcinfo;
    FunctionCallInfoData locfcinfo_sort;

    if (element_type != ARR_ELEMTYPE(array2))
        ereport(
            ERROR, (errcode(ERRCODE_DATATYPE_MISMATCH), errmsg("cannot compare arrays of different element types")));

    /* fast path if the arrays do not have the same dimensionality */
    if (ndims1 != ndims2 || memcmp(dims1, dims2, 2 * ndims1 * sizeof(int)) != 0) {
        result = false;
    } else {
        /*
         * We arrange to look up the equality function only once per series of
         * calls, assuming the element type doesn't change underneath us.  The
         * typcache is used so that we have no memory leakage when being used
         * as an index support function.
         */
        typentry = (TypeCacheEntry*)fcinfo->flinfo->fn_extra;
        typentry_sort = (TypeCacheEntry*)fcinfo->flinfo->fn_extra;
        if (typentry == NULL || typentry->type_id != element_type) {
            typentry = lookup_type_cache(element_type, TYPECACHE_EQ_OPR_FINFO);
            if (!OidIsValid(typentry->eq_opr_finfo.fn_oid))
                ereport(ERROR,
                    (errcode(ERRCODE_UNDEFINED_FUNCTION),
                        errmsg("could not identify a comparison operator for type %s", format_type_be(element_type))));
            fcinfo->flinfo->fn_extra = (void*)typentry;
        }
        typlen = typentry->typlen;
        typbyval = typentry->typbyval;
        typalign = typentry->typalign;

        if (typentry_sort == NULL || typentry_sort->type_id != element_type) {
            typentry_sort = lookup_type_cache(element_type, TYPECACHE_CMP_PROC_FINFO);
            if (!OidIsValid(typentry_sort->cmp_proc_finfo.fn_oid))
                ereport(ERROR,
                    (errcode(ERRCODE_UNDEFINED_FUNCTION),
                        errmsg("could not identify a comparison operator for type %s", format_type_be(element_type))));
        }

        /*
         * apply the operator to each pair of array elements.
         */
        InitFunctionCallInfoData(locfcinfo, &typentry->eq_opr_finfo, TWO_ARGS, collation, NULL, NULL);

        /* Loop over source data */
        int nitems = ArrayGetNItems(ndims1, dims1);
        
        ArrayItem* arr1 = (ArrayItem*)palloc0(nitems * sizeof(ArrayItem));
        ArrayItem* arr2 = (ArrayItem*)palloc0(nitems * sizeof(ArrayItem));

        char* ptr1 = ARR_DATA_PTR(array1);
        char* ptr2 = ARR_DATA_PTR(array2);
        bits8* bitmap1 = ARR_NULLBITMAP(array1);
        bits8* bitmap2 = ARR_NULLBITMAP(array2);
        uint32 bitmask = 1; /* use same bitmask for both arrays */

        for (int i = 0; i < nitems; i++) {
            Datum elt1;
            Datum elt2;
            bool isnull1 = bitmap1 && (*bitmap1 & bitmask) == 0;
            bool isnull2 = bitmap2 && (*bitmap2 & bitmask) == 0;

            /* In other databases, arrays containing null values will yield unequal results */
            if (isnull1 || isnull2) {
                pfree_ext(arr1);
                pfree_ext(arr2);
                PG_FREE_IF_COPY(array1, 0);
                PG_FREE_IF_COPY(array2, 1);
                return false;
            }

            /* Get elements, checking for NULL */
            inner_init_element(isnull1, &elt1, &ptr1, &arr1[i], typlen, typbyval, typalign);
            inner_init_element(isnull2, &elt2, &ptr2, &arr2[i], typlen, typbyval, typalign);
            arr1[i].elt = elt1;
            arr2[i].elt = elt2;

            /* advance bitmap pointers if any */
            bitmask <<= 1;
            if (bitmask == 0x100) {
                if (bitmap1 != NULL) {
                    bitmap1++;
                }
                if (bitmap2 != NULL) {
                    bitmap2++;
                }
                bitmask = 1;
            }
        }

        CmpFuncArgs* cmpFuncArgs = (CmpFuncArgs*)palloc0(sizeof(CmpFuncArgs));
        cmpFuncArgs->func = typentry_sort->cmp_proc_finfo;
        cmpFuncArgs->locfcinfo = locfcinfo_sort;
        cmpFuncArgs->collation = collation;
        qsort_arg(arr1, nitems, sizeof(ArrayItem), CmpArrayItem, cmpFuncArgs);
        qsort_arg(arr2, nitems, sizeof(ArrayItem), CmpArrayItem, cmpFuncArgs);
        pfree(cmpFuncArgs);

        for (int j = 0; j < nitems; j++) {
            ArrayItem elt1 = arr1[j];
            ArrayItem elt2 = arr2[j];
            bool oprresult = false;

            /*
             * We consider two NULLs equal; NULL and not-NULL are unequal.
             */
            if (elt1.isnull && elt2.isnull) {
                continue;
            }
            if (elt1.isnull || elt2.isnull) {
                result = false;
                break;
            }

            /*
             * Apply the operator to the element pair
             */
            locfcinfo.arg[0] = elt1.elt;
            locfcinfo.arg[1] = elt2.elt;
            locfcinfo.argnull[0] = false;
            locfcinfo.argnull[1] = false;
            locfcinfo.isnull = false;
            oprresult = DatumGetBool(FunctionCallInvoke(&locfcinfo));
            if (!oprresult) {
                result = false;
                break;
            }
        }
        pfree_ext(arr1);
        pfree_ext(arr2);
    }

    /* Avoid leaking memory when handed toasted input. */
    PG_FREE_IF_COPY(array1, 0);
    PG_FREE_IF_COPY(array2, 1);

    return result;
}

/*
 * array_eq :
 *		  compares two arrays for equality
 * result :
 *		  returns true if the arrays are equal, false otherwise.
 *
 * Note: we do not use array_cmp here, since equality may be meaningful in
 * datatypes that don't have a total ordering (and hence no btree support).
 */
Datum array_eq(PG_FUNCTION_ARGS)
{
    ArrayType* array1 = PG_GETARG_ARRAYTYPE_P(0);
    ArrayType* array2 = PG_GETARG_ARRAYTYPE_P(1);
    Oid collation = PG_GET_COLLATION();
    bool result = true;

    if (DB_IS_CMPT(A_FORMAT) && COMPATIBLE_A_DB_ARRAY) {
        result = array_eq_db_a_inner(array1, array2, collation, fcinfo);
    } else {
        result = array_eq_inner(array1, array2, collation, fcinfo);
    }

    PG_RETURN_BOOL(result);
}

/* -----------------------------------------------------------------------------
 * array-array bool operators:
 *		Given two arrays, iterate comparison operators
 *		over the array. Uses logic similar to text comparison
 *		functions, except element-by-element instead of
 *		character-by-character.
 * ----------------------------------------------------------------------------
 */

Datum array_ne(PG_FUNCTION_ARGS)
{
    PG_RETURN_BOOL(!DatumGetBool(array_eq(fcinfo)));
}

Datum array_lt(PG_FUNCTION_ARGS)
{
    PG_RETURN_BOOL(array_cmp(fcinfo) < 0);
}

Datum array_gt(PG_FUNCTION_ARGS)
{
    PG_RETURN_BOOL(array_cmp(fcinfo) > 0);
}

Datum array_le(PG_FUNCTION_ARGS)
{
    PG_RETURN_BOOL(array_cmp(fcinfo) <= 0);
}

Datum array_ge(PG_FUNCTION_ARGS)
{
    PG_RETURN_BOOL(array_cmp(fcinfo) >= 0);
}

Datum btarraycmp(PG_FUNCTION_ARGS)
{
    PG_RETURN_INT32(array_cmp(fcinfo));
}

/*
 * array_cmp()
 * Internal comparison function for arrays.
 *
 * Returns -1, 0 or 1
 */
static int array_cmp(FunctionCallInfo fcinfo)
{
    ArrayType* array1 = PG_GETARG_ARRAYTYPE_P(0);
    ArrayType* array2 = PG_GETARG_ARRAYTYPE_P(1);
    Oid collation = PG_GET_COLLATION();
    int ndims1 = ARR_NDIM(array1);
    int ndims2 = ARR_NDIM(array2);
    int* dims1 = ARR_DIMS(array1);
    int* dims2 = ARR_DIMS(array2);
    int nitems1 = ArrayGetNItems(ndims1, dims1);
    int nitems2 = ArrayGetNItems(ndims2, dims2);
    Oid element_type = ARR_ELEMTYPE(array1);
    int result = 0;
    TypeCacheEntry* typentry = NULL;
    int typlen;
    bool typbyval = false;
    char typalign;
    int min_nitems;
    char* ptr1 = NULL;
    char* ptr2 = NULL;
    bits8* bitmap1 = NULL;
    bits8* bitmap2 = NULL;
    uint32 bitmask;
    int i;
    FunctionCallInfoData locfcinfo;

    if (element_type != ARR_ELEMTYPE(array2))
        ereport(
            ERROR, (errcode(ERRCODE_DATATYPE_MISMATCH), errmsg("cannot compare arrays of different element types")));

    /*
     * We arrange to look up the comparison function only once per series of
     * calls, assuming the element type doesn't change underneath us. The
     * typcache is used so that we have no memory leakage when being used as
     * an index support function.
     */
    typentry = (TypeCacheEntry*)fcinfo->flinfo->fn_extra;
    if (typentry == NULL || typentry->type_id != element_type) {
        typentry = lookup_type_cache(element_type, TYPECACHE_CMP_PROC_FINFO);
        if (!OidIsValid(typentry->cmp_proc_finfo.fn_oid))
            ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_FUNCTION),
                    errmsg("could not identify a comparison function for type %s", format_type_be(element_type))));
        fcinfo->flinfo->fn_extra = (void*)typentry;
    }
    typlen = typentry->typlen;
    typbyval = typentry->typbyval;
    typalign = typentry->typalign;

    /*
     * apply the operator to each pair of array elements.
     */
    InitFunctionCallInfoData(locfcinfo, &typentry->cmp_proc_finfo, 2, collation, NULL, NULL);

    /* Loop over source data */
    min_nitems = Min(nitems1, nitems2);
    ptr1 = ARR_DATA_PTR(array1);
    ptr2 = ARR_DATA_PTR(array2);
    bitmap1 = ARR_NULLBITMAP(array1);
    bitmap2 = ARR_NULLBITMAP(array2);
    bitmask = 1; /* use same bitmask for both arrays */

    for (i = 0; i < min_nitems; i++) {
        Datum elt1;
        Datum elt2;
        bool isnull1 = false;
        bool isnull2 = false;
        int32 cmpresult;

        /* Get elements, checking for NULL */
        if (bitmap1 && (*bitmap1 & bitmask) == 0) {
            isnull1 = true;
            elt1 = (Datum)0;
        } else {
            isnull1 = false;
            elt1 = fetch_att(ptr1, typbyval, typlen);
            ptr1 = att_addlength_pointer(ptr1, typlen, ptr1);
            ptr1 = (char*)att_align_nominal(ptr1, typalign);
        }

        if (bitmap2 && (*bitmap2 & bitmask) == 0) {
            isnull2 = true;
            elt2 = (Datum)0;
        } else {
            isnull2 = false;
            elt2 = fetch_att(ptr2, typbyval, typlen);
            ptr2 = att_addlength_pointer(ptr2, typlen, ptr2);
            ptr2 = (char*)att_align_nominal(ptr2, typalign);
        }

        /* advance bitmap pointers if any */
        bitmask <<= 1;
        if (bitmask == 0x100) {
            if (bitmap1 != NULL)
                bitmap1++;
            if (bitmap2 != NULL)
                bitmap2++;
            bitmask = 1;
        }

        /*
         * We consider two NULLs equal; NULL > not-NULL.
         */
        if (isnull1 && isnull2)
            continue;
        if (isnull1) {
            /* arg1 is greater than arg2 */
            result = 1;
            break;
        }
        if (isnull2) {
            /* arg1 is less than arg2 */
            result = -1;
            break;
        }

        /* Compare the pair of elements */
        locfcinfo.arg[0] = elt1;
        locfcinfo.arg[1] = elt2;
        locfcinfo.argnull[0] = false;
        locfcinfo.argnull[1] = false;
        locfcinfo.isnull = false;
        cmpresult = DatumGetInt32(FunctionCallInvoke(&locfcinfo));
        if (cmpresult == 0)
            continue; /* equal */

        if (cmpresult < 0) {
            /* arg1 is less than arg2 */
            result = -1;
            break;
        } else {
            /* arg1 is greater than arg2 */
            result = 1;
            break;
        }
    }

    /*
     * If arrays contain same data (up to end of shorter one), apply
     * additional rules to sort by dimensionality.	The relative significance
     * of the different bits of information is historical; mainly we just care
     * that we don't say "equal" for arrays of different dimensionality.
     */
    if (result == 0) {
        if (nitems1 != nitems2)
            result = (nitems1 < nitems2) ? -1 : 1;
        else if (ndims1 != ndims2)
            result = (ndims1 < ndims2) ? -1 : 1;
        else {
            /* this relies on LB array immediately following DIMS array */
            for (i = 0; i < ndims1 * 2; i++) {
                if (dims1[i] != dims2[i]) {
                    result = (dims1[i] < dims2[i]) ? -1 : 1;
                    break;
                }
            }
        }
    }

    /* Avoid leaking memory when handed toasted input. */
    PG_FREE_IF_COPY(array1, 0);
    PG_FREE_IF_COPY(array2, 1);

    return result;
}

/* -----------------------------------------------------------------------------
 * array hashing
 *		Hash the elements and combine the results.
 * ----------------------------------------------------------------------------
 */

Datum hash_array(PG_FUNCTION_ARGS)
{
    ArrayType* array = PG_GETARG_ARRAYTYPE_P(0);
    int ndims = ARR_NDIM(array);
    int* dims = ARR_DIMS(array);
    Oid element_type = ARR_ELEMTYPE(array);
    uint32 result = 1;
    int nitems;
    TypeCacheEntry* typentry = NULL;
    int typlen;
    bool typbyval = false;
    char typalign;
    char* ptr = NULL;
    bits8* bitmap = NULL;
    uint32 bitmask;
    int i;
    FunctionCallInfoData locfcinfo;

    /*
     * We arrange to look up the hash function only once per series of calls,
     * assuming the element type doesn't change underneath us.  The typcache
     * is used so that we have no memory leakage when being used as an index
     * support function.
     */
    typentry = (TypeCacheEntry*)fcinfo->flinfo->fn_extra;
    if (typentry == NULL || typentry->type_id != element_type) {
        typentry = lookup_type_cache(element_type, TYPECACHE_HASH_PROC_FINFO);
        if (!OidIsValid(typentry->hash_proc_finfo.fn_oid))
            ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_FUNCTION),
                    errmsg("could not identify a hash function for type %s", format_type_be(element_type))));
        fcinfo->flinfo->fn_extra = (void*)typentry;
    }
    typlen = typentry->typlen;
    typbyval = typentry->typbyval;
    typalign = typentry->typalign;

    /*
     * apply the hash function to each array element.
     */
    InitFunctionCallInfoData(locfcinfo, &typentry->hash_proc_finfo, 1, InvalidOid, NULL, NULL);

    /* Loop over source data */
    nitems = ArrayGetNItems(ndims, dims);
    ptr = ARR_DATA_PTR(array);
    bitmap = ARR_NULLBITMAP(array);
    bitmask = 1;

    for (i = 0; i < nitems; i++) {
        uint32 elthash;

        /* Get element, checking for NULL */
        if (bitmap && (*bitmap & bitmask) == 0) {
            /* Treat nulls as having hashvalue 0 */
            elthash = 0;
        } else {
            Datum elt;

            elt = fetch_att(ptr, typbyval, typlen);
            ptr = att_addlength_pointer(ptr, typlen, ptr);
            ptr = (char*)att_align_nominal(ptr, typalign);

            /* Apply the hash function */
            locfcinfo.arg[0] = elt;
            locfcinfo.argnull[0] = false;
            locfcinfo.isnull = false;
            elthash = DatumGetUInt32(FunctionCallInvoke(&locfcinfo));
        }

        /* advance bitmap pointer if any */
        if (bitmap != NULL) {
            bitmask <<= 1;
            if (bitmask == 0x100) {
                bitmap++;
                bitmask = 1;
            }
        }

        /*
         * Combine hash values of successive elements by multiplying the
         * current value by 31 and adding on the new element's hash value.
         *
         * The result is a sum in which each element's hash value is
         * multiplied by a different power of 31. This is modulo 2^32
         * arithmetic, and the powers of 31 modulo 2^32 form a cyclic group of
         * order 2^27. So for arrays of up to 2^27 elements, each element's
         * hash value is multiplied by a different (odd) number, resulting in
         * a good mixing of all the elements' hash values.
         */
        result = (result << 5) - result + elthash;
    }

    /* Avoid leaking memory when handed toasted input. */
    PG_FREE_IF_COPY(array, 0);

    PG_RETURN_UINT32(result);
}

/* -----------------------------------------------------------------------------
 * array overlap/containment comparisons
 *		These use the same methods of comparing array elements as array_eq.
 *		We consider only the elements of the arrays, ignoring dimensionality.
 * ----------------------------------------------------------------------------
 */

/*
 * array_contain_compare :
 *		  compares two arrays for overlap/containment
 *
 * When matchall is true, return true if all members of array1 are in array2.
 * When matchall is false, return true if any members of array1 are in array2.
 */
static bool array_contain_compare(ArrayType* array1, ArrayType* array2, Oid collation, bool matchall, void** fn_extra)
{
    bool result = matchall;
    Oid element_type = ARR_ELEMTYPE(array1);
    TypeCacheEntry* typentry = NULL;
    int nelems1;
    Datum* values2 = NULL;
    bool* nulls2 = NULL;
    int nelems2;
    int typlen;
    bool typbyval = false;
    char typalign;
    char* ptr1 = NULL;
    bits8* bitmap1 = NULL;
    uint32 bitmask;
    int i;
    int j;
    FunctionCallInfoData locfcinfo;

    if (element_type != ARR_ELEMTYPE(array2))
        ereport(
            ERROR, (errcode(ERRCODE_DATATYPE_MISMATCH), errmsg("cannot compare arrays of different element types")));

    /*
     * We arrange to look up the equality function only once per series of
     * calls, assuming the element type doesn't change underneath us.  The
     * typcache is used so that we have no memory leakage when being used as
     * an index support function.
     */
    typentry = (TypeCacheEntry*)*fn_extra;
    if (typentry == NULL || typentry->type_id != element_type) {
        typentry = lookup_type_cache(element_type, TYPECACHE_EQ_OPR_FINFO);
        if (!OidIsValid(typentry->eq_opr_finfo.fn_oid))
            ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_FUNCTION),
                    errmsg("could not identify an equality operator for type %s", format_type_be(element_type))));
        *fn_extra = (void*)typentry;
    }
    typlen = typentry->typlen;
    typbyval = typentry->typbyval;
    typalign = typentry->typalign;

    /*
     * Since we probably will need to scan array2 multiple times, it's
     * worthwhile to use deconstruct_array on it.  We scan array1 the hard way
     * however, since we very likely won't need to look at all of it.
     */
    deconstruct_array(array2, element_type, typlen, typbyval, typalign, &values2, &nulls2, &nelems2);

    /*
     * Apply the comparison operator to each pair of array elements.
     */
    InitFunctionCallInfoData(locfcinfo, &typentry->eq_opr_finfo, 2, collation, NULL, NULL);

    /* Loop over source data */
    nelems1 = ArrayGetNItems(ARR_NDIM(array1), ARR_DIMS(array1));
    ptr1 = ARR_DATA_PTR(array1);
    bitmap1 = ARR_NULLBITMAP(array1);
    bitmask = 1;

    for (i = 0; i < nelems1; i++) {
        Datum elt1;
        bool isnull1 = false;

        /* Get element, checking for NULL */
        if (bitmap1 && (*bitmap1 & bitmask) == 0) {
            isnull1 = true;
            elt1 = (Datum)0;
        } else {
            isnull1 = false;
            elt1 = fetch_att(ptr1, typbyval, typlen);
            ptr1 = att_addlength_pointer(ptr1, typlen, ptr1);
            ptr1 = (char*)att_align_nominal(ptr1, typalign);
        }

        /* advance bitmap pointer if any */
        bitmask <<= 1;
        if (bitmask == 0x100) {
            if (bitmap1 != NULL) {
                bitmap1++;
            }
            bitmask = 1;
        }

        /*
         * We assume that the comparison operator is strict, so a NULL can't
         * match anything.	XXX this diverges from the "NULL=NULL" behavior of
         * array_eq, should we act like that?
         */
        if (isnull1) {
            if (matchall) {
                result = false;
                break;
            }
            continue;
        }

        for (j = 0; j < nelems2; j++) {
            Datum elt2 = values2[j];
            bool isnull2 = nulls2[j];
            bool oprresult = false;

            if (isnull2)
                continue; /* can't match */

            /*
             * Apply the operator to the element pair
             */
            locfcinfo.arg[0] = elt1;
            locfcinfo.arg[1] = elt2;
            locfcinfo.argnull[0] = false;
            locfcinfo.argnull[1] = false;
            locfcinfo.isnull = false;
            oprresult = DatumGetBool(FunctionCallInvoke(&locfcinfo));
            if (oprresult)
                break;
        }

        if (j < nelems2) {
            /* found a match for elt1 */
            if (!matchall) {
                result = true;
                break;
            }
        } else {
            /* no match for elt1 */
            if (matchall) {
                result = false;
                break;
            }
        }
    }

    pfree(values2);
    pfree(nulls2);

    return result;
}

Datum arrayoverlap(PG_FUNCTION_ARGS)
{
    ArrayType* array1 = PG_GETARG_ARRAYTYPE_P(0);
    ArrayType* array2 = PG_GETARG_ARRAYTYPE_P(1);
    Oid collation = PG_GET_COLLATION();
    bool result = false;

    result = array_contain_compare(array1, array2, collation, false, &fcinfo->flinfo->fn_extra);

    /* Avoid leaking memory when handed toasted input. */
    PG_FREE_IF_COPY(array1, 0);
    PG_FREE_IF_COPY(array2, 1);

    PG_RETURN_BOOL(result);
}

Datum arraycontains(PG_FUNCTION_ARGS)
{
    ArrayType* array1 = PG_GETARG_ARRAYTYPE_P(0);
    ArrayType* array2 = PG_GETARG_ARRAYTYPE_P(1);
    Oid collation = PG_GET_COLLATION();
    bool result = false;

    result = array_contain_compare(array2, array1, collation, true, &fcinfo->flinfo->fn_extra);

    /* Avoid leaking memory when handed toasted input. */
    PG_FREE_IF_COPY(array1, 0);
    PG_FREE_IF_COPY(array2, 1);

    PG_RETURN_BOOL(result);
}

Datum arraycontained(PG_FUNCTION_ARGS)
{
    ArrayType* array1 = PG_GETARG_ARRAYTYPE_P(0);
    ArrayType* array2 = PG_GETARG_ARRAYTYPE_P(1);
    Oid collation = PG_GET_COLLATION();
    bool result = false;

    result = array_contain_compare(array1, array2, collation, true, &fcinfo->flinfo->fn_extra);

    /* Avoid leaking memory when handed toasted input. */
    PG_FREE_IF_COPY(array1, 0);
    PG_FREE_IF_COPY(array2, 1);

    PG_RETURN_BOOL(result);
}

/* -----------------------------------------------------------------------------
 * Array iteration functions
 *		These functions are used to iterate efficiently through arrays
 * -----------------------------------------------------------------------------
 */

/*
 * array_create_iterator --- set up to iterate through an array
 *
 * If slice_ndim is zero, we will iterate element-by-element; the returned
 * datums are of the array's element type.
 *
 * If slice_ndim is 1..ARR_NDIM(arr), we will iterate by slices: the
 * returned datums are of the same array type as 'arr', but of size
 * equal to the rightmost N dimensions of 'arr'.
 *
 * The passed-in array must remain valid for the lifetime of the iterator.
 */
ArrayIterator array_create_iterator(ArrayType* arr, int slice_ndim, ArrayMetaState* mstate)
{
    ArrayIterator iterator = (ArrayIteratorData*)palloc0(sizeof(ArrayIteratorData));

    /*
     * Sanity-check inputs --- caller should have got this right already
     */
    Assert(PointerIsValid(arr));
    if (slice_ndim < 0 || slice_ndim > ARR_NDIM(arr))
        ereport(ERROR, (errcode(ERRCODE_ARRAY_ELEMENT_ERROR), errmsg("invalid arguments to array_create_iterator")));

    /*
     * Remember basic info about the array and its element type
     */
    iterator->arr = arr;
    iterator->nullbitmap = ARR_NULLBITMAP(arr);
    iterator->nitems = ArrayGetNItems(ARR_NDIM(arr), ARR_DIMS(arr));
    if (mstate != NULL) {
        Assert(mstate->element_type == ARR_ELEMTYPE(arr));

        iterator->typlen = mstate->typlen;
        iterator->typbyval = mstate->typbyval;
        iterator->typalign = mstate->typalign;
    } else {
        get_typlenbyvalalign(ARR_ELEMTYPE(arr), &iterator->typlen, &iterator->typbyval, &iterator->typalign);
    }

    /*
     * Remember the slicing parameters.
     */
    iterator->slice_ndim = slice_ndim;

    if (slice_ndim > 0) {
        /*
         * Get pointers into the array's dims and lbound arrays to represent
         * the dims/lbound arrays of a slice.  These are the same as the
         * rightmost N dimensions of the array.
         */
        iterator->slice_dims = ARR_DIMS(arr) + ARR_NDIM(arr) - slice_ndim;
        iterator->slice_lbound = ARR_LBOUND(arr) + ARR_NDIM(arr) - slice_ndim;

        /*
         * Compute number of elements in a slice.
         */
        iterator->slice_len = ArrayGetNItems(slice_ndim, iterator->slice_dims);

        /*
         * Create workspace for building sub-arrays.
         */
        iterator->slice_values = (Datum*)palloc(iterator->slice_len * sizeof(Datum));
        iterator->slice_nulls = (bool*)palloc(iterator->slice_len * sizeof(bool));
    }

    /*
     * Initialize our data pointer and linear element number.  These will
     * advance through the array during array_iterate().
     */
    iterator->data_ptr = ARR_DATA_PTR(arr);
    iterator->current_item = 0;

    return iterator;
}

/*
 * Iterate through the array referenced by 'iterator'.
 *
 * As long as there is another element (or slice), return it into
 * *value / *isnull, and return true.  Return false when no more data.
 */
bool array_iterate(ArrayIterator iterator, Datum* value, bool* isnull)
{
    /* Done if we have reached the end of the array */
    if (iterator->current_item >= iterator->nitems)
        return false;

    if (iterator->slice_ndim == 0) {
        /*
         * Scalar case: return one element.
         */
        if (array_get_isnull(iterator->nullbitmap, iterator->current_item++)) {
            *isnull = true;
            *value = (Datum)0;
        } else {
            /* non-NULL, so fetch the individual Datum to return */
            char* p = iterator->data_ptr;

            *isnull = false;
            *value = fetch_att(p, iterator->typbyval, iterator->typlen);

            /* Move our data pointer forward to the next element */
            p = att_addlength_pointer(p, iterator->typlen, p);
            p = (char*)att_align_nominal(p, iterator->typalign);
            iterator->data_ptr = p;
        }
    } else {
        /*
         * Slice case: build and return an array of the requested size.
         */
        ArrayType* result = NULL;
        Datum* values = iterator->slice_values;
        bool* nulls = iterator->slice_nulls;
        char* p = iterator->data_ptr;
        int i;

        for (i = 0; i < iterator->slice_len; i++) {
            if (array_get_isnull(iterator->nullbitmap, iterator->current_item++)) {
                nulls[i] = true;
                values[i] = (Datum)0;
            } else {
                nulls[i] = false;
                values[i] = fetch_att(p, iterator->typbyval, iterator->typlen);

                /* Move our data pointer forward to the next element */
                p = att_addlength_pointer(p, iterator->typlen, p);
                p = (char*)att_align_nominal(p, iterator->typalign);
            }
        }

        iterator->data_ptr = p;

        result = construct_md_array(values,
            nulls,
            iterator->slice_ndim,
            iterator->slice_dims,
            iterator->slice_lbound,
            ARR_ELEMTYPE(iterator->arr),
            iterator->typlen,
            iterator->typbyval,
            iterator->typalign);

        *isnull = false;
        *value = PointerGetDatum(result);
    }

    return true;
}

/*
 * Release an ArrayIterator data structure
 */
void array_free_iterator(ArrayIterator iterator)
{
    if (iterator->slice_ndim > 0) {
        pfree(iterator->slice_values);
        pfree(iterator->slice_nulls);
    }
    pfree(iterator);
}

/***************************************************************************/
/******************|		  Support  Routines			  |*****************/
/***************************************************************************/

/*
 * Check whether a specific array element is NULL
 *
 * nullbitmap: pointer to array's null bitmap (NULL if none)
 * offset: 0-based linear element number of array element
 */
static bool array_get_isnull(const bits8* nullbitmap, int offset)
{
    if (nullbitmap == NULL)
        return false; /* assume not null */
    if (nullbitmap[offset / 8] & (1 << (offset % 8)))
        return false; /* not null */
    return true;
}

/*
 * Set a specific array element's null-bitmap entry
 *
 * nullbitmap: pointer to array's null bitmap (mustn't be NULL)
 * offset: 0-based linear element number of array element
 * isNull: null status to set
 */
static void array_set_isnull(bits8* nullbitmap, int offset, bool isNull)
{
    uint32 bitmask;

    nullbitmap += offset / 8;
    bitmask = 1 << (offset % 8);
    if (isNull)
        *nullbitmap &= ~bitmask;
    else
        *nullbitmap |= bitmask;
}

/*
 * Fetch array element at pointer, converted correctly to a Datum
 *
 * Caller must have handled case of NULL element
 */
static Datum ArrayCast(const char* value, bool byval, int len)
{
    return fetch_att(value, byval, len);
}

/*
 * Copy datum to *dest and return total space used (including align padding)
 *
 * Caller must have handled case of NULL element
 */
int ArrayCastAndSet(Datum src, int typlen, bool typbyval, char typalign, char* dest)
{
    int inc;

    if (typlen > 0) {
        if (typbyval)
            store_att_byval(dest, src, typlen);
        else {
            errno_t rc = memmove_s(dest, typlen, DatumGetPointer(src), typlen);
            securec_check(rc, "\0", "\0");
        }
        inc = att_align_nominal(typlen, typalign);
    } else {
        Assert(!typbyval);
        inc = att_addlength_datum(0, typlen, src);
        errno_t rc = memmove_s(dest, inc, DatumGetPointer(src), inc);
        securec_check(rc, "\0", "\0");
        inc = att_align_nominal(inc, typalign);
    }

    return inc;
}

/*
 * Advance ptr over nitems array elements
 *
 * ptr: starting location in array
 * offset: 0-based linear element number of first element (the one at *ptr)
 * nullbitmap: start of array's null bitmap, or NULL if none
 * nitems: number of array elements to advance over (>= 0)
 * typlen, typbyval, typalign: storage parameters of array element datatype
 *
 * It is caller's responsibility to ensure that nitems is within range
 */
static char* array_seek(char* ptr, int offset, bits8* nullbitmap, int nitems, int typlen, bool typbyval, char typalign)
{
    uint32 bitmask;
    int i;

    /* easy if fixed-size elements and no NULLs */
    if (typlen > 0 && nullbitmap == NULL)
        return ptr + nitems * ((Size)att_align_nominal(typlen, typalign));

    /* seems worth having separate loops for NULL and no-NULLs cases */
    if (nullbitmap != NULL) {
        nullbitmap += offset / 8;
        bitmask = 1 << (offset % 8);

        for (i = 0; i < nitems; i++) {
            if (*nullbitmap & bitmask) {
                ptr = att_addlength_pointer(ptr, typlen, ptr);
                ptr = (char*)att_align_nominal(ptr, typalign);
            }
            bitmask <<= 1;
            if (bitmask == 0x100) {
                nullbitmap++;
                bitmask = 1;
            }
        }
    } else {
        for (i = 0; i < nitems; i++) {
            ptr = att_addlength_pointer(ptr, typlen, ptr);
            ptr = (char*)att_align_nominal(ptr, typalign);
        }
    }
    return ptr;
}

/*
 * Compute total size of the nitems array elements starting at *ptr
 *
 * Parameters same as for array_seek
 */
static int array_nelems_size(
    char* ptr, int offset, bits8* nullbitmap, int nitems, int typlen, bool typbyval, char typalign)
{
    return array_seek(ptr, offset, nullbitmap, nitems, typlen, typbyval, typalign) - ptr;
}

/*
 * Copy nitems array elements from srcptr to destptr
 *
 * destptr: starting destination location (must be enough room!)
 * nitems: number of array elements to copy (>= 0)
 * srcptr: starting location in source array
 * offset: 0-based linear element number of first element (the one at *srcptr)
 * nullbitmap: start of source array's null bitmap, or NULL if none
 * typlen, typbyval, typalign: storage parameters of array element datatype
 *
 * Returns number of bytes copied
 *
 * NB: this does not take care of setting up the destination's null bitmap!
 */
static int array_copy(
    char* destptr, int nitems, char* srcptr, int offset, bits8* nullbitmap, int typlen, bool typbyval, char typalign)
{
    int numbytes;

    numbytes = array_nelems_size(srcptr, offset, nullbitmap, nitems, typlen, typbyval, typalign);
    errno_t errorno = EOK;
    if (numbytes > 0) {
        errorno = memcpy_s(destptr, numbytes, srcptr, numbytes);
        securec_check(errorno, "\0", "\0");
    }
    return numbytes;
}

/*
 * Copy nitems null-bitmap bits from source to destination
 *
 * destbitmap: start of destination array's null bitmap (mustn't be NULL)
 * destoffset: 0-based linear element number of first dest element
 * srcbitmap: start of source array's null bitmap, or NULL if none
 * srcoffset: 0-based linear element number of first source element
 * nitems: number of bits to copy (>= 0)
 *
 * If srcbitmap is NULL then we assume the source is all-non-NULL and
 * fill 1's into the destination bitmap.  Note that only the specified
 * bits in the destination map are changed, not any before or after.
 *
 * Note: this could certainly be optimized using standard bitblt methods.
 * However, it's not clear that the typical openGauss array has enough elements
 * to make it worth worrying too much.	For the moment, KISS.
 */
void array_bitmap_copy(bits8* destbitmap, int destoffset, const bits8* srcbitmap, int srcoffset, int nitems)
{
    uint32 destbitmask, destbitval, srcbitmask, srcbitval;

    Assert(destbitmap);
    if (nitems <= 0)
        return; /* don't risk fetch off end of memory */
    destbitmap += destoffset / 8;
    destbitmask = 1 << (destoffset % 8);
    destbitval = *destbitmap;
    if (srcbitmap != NULL) {
        srcbitmap += srcoffset / 8;
        srcbitmask = 1 << (srcoffset % 8);
        srcbitval = *srcbitmap;
        while (nitems-- > 0) {
            if (srcbitval & srcbitmask)
                destbitval |= destbitmask;
            else
                destbitval &= ~destbitmask;
            destbitmask <<= 1;
            if (destbitmask == 0x100) {
                *destbitmap++ = destbitval;
                destbitmask = 1;
                if (nitems > 0)
                    destbitval = *destbitmap;
            }
            srcbitmask <<= 1;
            if (srcbitmask == 0x100) {
                srcbitmap++;
                srcbitmask = 1;
                if (nitems > 0)
                    srcbitval = *srcbitmap;
            }
        }
        if (destbitmask != 1)
            *destbitmap = destbitval;
    } else {
        while (nitems-- > 0) {
            destbitval |= destbitmask;
            destbitmask <<= 1;
            if (destbitmask == 0x100) {
                *destbitmap++ = destbitval;
                destbitmask = 1;
                if (nitems > 0)
                    destbitval = *destbitmap;
            }
        }
        if (destbitmask != 1)
            *destbitmap = destbitval;
    }
}

/*
 * Compute space needed for a slice of an array
 *
 * We assume the caller has verified that the slice coordinates are valid.
 */
static int array_slice_size(char* arraydataptr, bits8* arraynullsptr, int ndim, int* dim, int* lb, const int* st,
    int* endp, int typlen, bool typbyval, char typalign)
{
    int src_offset, span[MAXDIM], prod[MAXDIM], dist[MAXDIM], indx[MAXDIM];
    char* ptr = NULL;
    int i, j, inc;
    int count = 0;

    mda_get_range(ndim, span, st, endp);

    /* Pretty easy for fixed element length without nulls ... */
    if (typlen > 0 && arraynullsptr == NULL)
        return ArrayGetNItems(ndim, span) * att_align_nominal(typlen, typalign);

    /* Else gotta do it the hard way */
    src_offset = ArrayGetOffset(ndim, dim, lb, st);
    ptr = array_seek(arraydataptr, 0, arraynullsptr, src_offset, typlen, typbyval, typalign);
    mda_get_prod(ndim, dim, prod);
    mda_get_offset_values(ndim, dist, prod, span);
    for (i = 0; i < ndim; i++)
        indx[i] = 0;
    j = ndim - 1;
    do {
        if (dist[j]) {
            ptr = array_seek(ptr, src_offset, arraynullsptr, dist[j], typlen, typbyval, typalign);
            src_offset += dist[j];
        }
        if (!array_get_isnull(arraynullsptr, src_offset)) {
            inc = att_addlength_pointer(0, typlen, ptr);
            inc = att_align_nominal(inc, typalign);
            ptr += inc;
            count += inc;
        }
        src_offset++;
    } while ((j = mda_next_tuple(ndim, indx, span)) != -1);
    return count;
}

/*
 * Extract a slice of an array into consecutive elements in the destination
 * array.
 *
 * We assume the caller has verified that the slice coordinates are valid,
 * allocated enough storage for the result, and initialized the header
 * of the new array.
 */
static void array_extract_slice(ArrayType* newarray, int ndim, int* dim, int* lb, char* arraydataptr,
    bits8* arraynullsptr, const int* st, int* endp, int typlen, bool typbyval, char typalign)
{
    char* destdataptr = ARR_DATA_PTR(newarray);
    bits8* destnullsptr = ARR_NULLBITMAP(newarray);
    char* srcdataptr = NULL;
    int src_offset, dest_offset, prod[MAXDIM], span[MAXDIM], dist[MAXDIM], indx[MAXDIM];
    int i, j, inc;

    src_offset = ArrayGetOffset(ndim, dim, lb, st);
    srcdataptr = array_seek(arraydataptr, 0, arraynullsptr, src_offset, typlen, typbyval, typalign);
    mda_get_prod(ndim, dim, prod);
    mda_get_range(ndim, span, st, endp);
    mda_get_offset_values(ndim, dist, prod, span);
    for (i = 0; i < ndim; i++)
        indx[i] = 0;
    dest_offset = 0;
    j = ndim - 1;
    do {
        if (dist[j]) {
            /* skip unwanted elements */
            srcdataptr = array_seek(srcdataptr, src_offset, arraynullsptr, dist[j], typlen, typbyval, typalign);
            src_offset += dist[j];
        }
        inc = array_copy(destdataptr, 1, srcdataptr, src_offset, arraynullsptr, typlen, typbyval, typalign);
        if (destnullsptr != NULL)
            array_bitmap_copy(destnullsptr, dest_offset, arraynullsptr, src_offset, 1);
        destdataptr += inc;
        srcdataptr += inc;
        src_offset++;
        dest_offset++;
    } while ((j = mda_next_tuple(ndim, indx, span)) != -1);
}

/*
 * Insert a slice into an array.
 *
 * ndim/dim[]/lb[] are dimensions of the original array.  A new array with
 * those same dimensions is to be constructed.	destArray must already
 * have been allocated and its header initialized.
 *
 * st[]/endp[] identify the slice to be replaced.  Elements within the slice
 * volume are taken from consecutive elements of the srcArray; elements
 * outside it are copied from origArray.
 *
 * We assume the caller has verified that the slice coordinates are valid.
 */
static void array_insert_slice(ArrayType* destArray, ArrayType* origArray, ArrayType* srcArray, int ndim, int* dim,
    int* lb, const int* st, int* endp, int typlen, bool typbyval, char typalign)
{
    char* destPtr = ARR_DATA_PTR(destArray);
    char* origPtr = ARR_DATA_PTR(origArray);
    char* srcPtr = ARR_DATA_PTR(srcArray);
    bits8* destBitmap = ARR_NULLBITMAP(destArray);
    bits8* origBitmap = ARR_NULLBITMAP(origArray);
    bits8* srcBitmap = ARR_NULLBITMAP(srcArray);
    int orignitems = ArrayGetNItems(ARR_NDIM(origArray), ARR_DIMS(origArray));
    int dest_offset, orig_offset, src_offset, prod[MAXDIM], span[MAXDIM], dist[MAXDIM], indx[MAXDIM];
    int i, j, inc;

    dest_offset = ArrayGetOffset(ndim, dim, lb, st);
    /* copy items before the slice start */
    inc = array_copy(destPtr, dest_offset, origPtr, 0, origBitmap, typlen, typbyval, typalign);
    destPtr += inc;
    origPtr += inc;
    if (destBitmap != NULL)
        array_bitmap_copy(destBitmap, 0, origBitmap, 0, dest_offset);
    orig_offset = dest_offset;
    mda_get_prod(ndim, dim, prod);
    mda_get_range(ndim, span, st, endp);
    mda_get_offset_values(ndim, dist, prod, span);
    for (i = 0; i < ndim; i++)
        indx[i] = 0;
    src_offset = 0;
    j = ndim - 1;
    do {
        /* Copy/advance over elements between here and next part of slice */
        if (dist[j]) {
            inc = array_copy(destPtr, dist[j], origPtr, orig_offset, origBitmap, typlen, typbyval, typalign);
            destPtr += inc;
            origPtr += inc;
            if (destBitmap != NULL)
                array_bitmap_copy(destBitmap, dest_offset, origBitmap, orig_offset, dist[j]);
            dest_offset += dist[j];
            orig_offset += dist[j];
        }
        /* Copy new element at this slice position */
        inc = array_copy(destPtr, 1, srcPtr, src_offset, srcBitmap, typlen, typbyval, typalign);
        if (destBitmap != NULL)
            array_bitmap_copy(destBitmap, dest_offset, srcBitmap, src_offset, 1);
        destPtr += inc;
        srcPtr += inc;
        dest_offset++;
        src_offset++;
        /* Advance over old element at this slice position */
        origPtr = array_seek(origPtr, orig_offset, origBitmap, 1, typlen, typbyval, typalign);
        orig_offset++;
    } while ((j = mda_next_tuple(ndim, indx, span)) != -1);

    /* don't miss any data at the end */
    array_copy(destPtr, orignitems - orig_offset, origPtr, orig_offset, origBitmap, typlen, typbyval, typalign);
    if (destBitmap != NULL)
        array_bitmap_copy(destBitmap, dest_offset, origBitmap, orig_offset, orignitems - orig_offset);
}

/*
 * accumArrayResult - accumulate one (more) Datum for an array result
 *
 *	astate is working state (NULL on first call)
 *	rcontext is where to keep working state
 */
ArrayBuildState* accumArrayResult(
    ArrayBuildState* astate, Datum dvalue, bool disnull, Oid element_type, MemoryContext rcontext)
{
    MemoryContext arr_context, oldcontext;

    if (astate == NULL) {
        /* First time through --- initialize */

        /* Make a temporary context to hold all the junk */
        arr_context = AllocSetContextCreate(rcontext,
            "accumArrayResult",
            ALLOCSET_DEFAULT_MINSIZE,
            ALLOCSET_DEFAULT_INITSIZE,
            ALLOCSET_DEFAULT_MAXSIZE);
        oldcontext = MemoryContextSwitchTo(arr_context);
        astate = (ArrayBuildState*)palloc(sizeof(ArrayBuildState));
        astate->mcontext = arr_context;
        astate->alen = 64; /* arbitrary starting array size */
        astate->dvalues = (Datum*)palloc(astate->alen * sizeof(Datum));
        astate->dnulls = (bool*)palloc(astate->alen * sizeof(bool));
        astate->nelems = 0;
        astate->element_type = element_type;
        get_typlenbyvalalign(element_type, &astate->typlen, &astate->typbyval, &astate->typalign);
    } else {
        oldcontext = MemoryContextSwitchTo(astate->mcontext);
        Assert(astate->element_type == element_type);
        /* enlarge dvalues[]/dnulls[] if needed */
        if (astate->nelems >= astate->alen) {
            astate->alen *= 2;
            astate->dvalues = (Datum*)repalloc(astate->dvalues, astate->alen * sizeof(Datum));
            astate->dnulls = (bool*)repalloc(astate->dnulls, astate->alen * sizeof(bool));
        }
    }

    /*
     * Ensure pass-by-ref stuff is copied into mcontext; and detoast it too if
     * it's varlena.  (You might think that detoasting is not needed here
     * because construct_md_array can detoast the array elements later.
     * However, we must not let construct_md_array modify the ArrayBuildState
     * because that would mean array_agg_finalfn damages its input, which is
     * verboten.  Also, this way frequently saves one copying step.)
     */
    if (!disnull && !astate->typbyval) {
        if (astate->typlen == -1)
            dvalue = PointerGetDatum(PG_DETOAST_DATUM_COPY(dvalue));
        else
            dvalue = datumCopy(dvalue, astate->typbyval, astate->typlen);
    }

    astate->dvalues[astate->nelems] = dvalue;
    astate->dnulls[astate->nelems] = disnull;
    astate->nelems++;

    MemoryContextSwitchTo(oldcontext);

    return astate;
}

/*
 * makeArrayResult - produce 1-D final result of accumArrayResult
 *
 *	astate is working state (not NULL)
 *	rcontext is where to construct result
 */
Datum makeArrayResult(ArrayBuildState* astate, MemoryContext rcontext)
{
    int dims[1];
    int lbs[1];

    dims[0] = astate->nelems;
    lbs[0] = 1;

    return makeMdArrayResult(astate, 1, dims, lbs, rcontext, true);
}

/*
 * makeMdArrayResult - produce multi-D final result of accumArrayResult
 *
 * beware: no check that specified dimensions match the number of values
 * accumulated.
 *
 *	astate is working state (not NULL)
 *	rcontext is where to construct result
 *	release is true if okay to release working state
 */
Datum makeMdArrayResult(ArrayBuildState* astate, int ndims, int* dims, int* lbs, MemoryContext rcontext, bool release)
{
    ArrayType* result = NULL;
    MemoryContext oldcontext;

    /* Build the final array result in rcontext */
    oldcontext = MemoryContextSwitchTo(rcontext);

    result = construct_md_array(astate->dvalues,
        astate->dnulls,
        ndims,
        dims,
        lbs,
        astate->element_type,
        astate->typlen,
        astate->typbyval,
        astate->typalign);

    MemoryContextSwitchTo(oldcontext);

    /* Clean up all the junk */
    if (release)
        MemoryContextDelete(astate->mcontext);

    return PointerGetDatum(result);
}

Datum array_larger(PG_FUNCTION_ARGS)
{
    ArrayType *v1 = NULL, *v2 = NULL, *result = NULL;

    v1 = PG_GETARG_ARRAYTYPE_P(0);
    v2 = PG_GETARG_ARRAYTYPE_P(1);

    result = ((array_cmp(fcinfo) > 0) ? v1 : v2);

    PG_RETURN_ARRAYTYPE_P(result);
}

Datum array_smaller(PG_FUNCTION_ARGS)
{
    ArrayType *v1 = NULL, *v2 = NULL, *result = NULL;

    v1 = PG_GETARG_ARRAYTYPE_P(0);
    v2 = PG_GETARG_ARRAYTYPE_P(1);

    result = ((array_cmp(fcinfo) < 0) ? v1 : v2);

    PG_RETURN_ARRAYTYPE_P(result);
}

typedef struct generate_subscripts_fctx {
    int4 lower;
    int4 upper;
    bool reverse;
} generate_subscripts_fctx;

/*
 * generate_subscripts(array anyarray, dim int [, reverse bool])
 *		Returns all subscripts of the array for any dimension
 */
Datum generate_subscripts(PG_FUNCTION_ARGS)
{
    FuncCallContext* funcctx = NULL;
    MemoryContext oldcontext;
    generate_subscripts_fctx* fctx = NULL;

    /* stuff done only on the first call of the function */
    if (SRF_IS_FIRSTCALL()) {
        ArrayType* v = PG_GETARG_ARRAYTYPE_P(0);
        int reqdim = PG_GETARG_INT32(1);
        int *lb = NULL, *dimv = NULL;

        /* create a function context for cross-call persistence */
        funcctx = SRF_FIRSTCALL_INIT();

        /* Sanity check: does it look like an array at all? */
        if (ARR_NDIM(v) <= 0 || ARR_NDIM(v) > MAXDIM)
            SRF_RETURN_DONE(funcctx);

        /* Sanity check: was the requested dim valid */
        if (reqdim <= 0 || reqdim > ARR_NDIM(v))
            SRF_RETURN_DONE(funcctx);

        /*
         * switch to memory context appropriate for multiple function calls
         */
        oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
        fctx = (generate_subscripts_fctx*)palloc(sizeof(generate_subscripts_fctx));

        lb = ARR_LBOUND(v);
        dimv = ARR_DIMS(v);

        fctx->lower = lb[reqdim - 1];
        fctx->upper = dimv[reqdim - 1] + lb[reqdim - 1] - 1;
        fctx->reverse = (PG_NARGS() < 3) ? false : PG_GETARG_BOOL(2);

        funcctx->user_fctx = fctx;

        MemoryContextSwitchTo(oldcontext);
    }

    funcctx = SRF_PERCALL_SETUP();

    fctx = (generate_subscripts_fctx*)funcctx->user_fctx;

    if (fctx->lower <= fctx->upper) {
        if (!fctx->reverse)
            SRF_RETURN_NEXT(funcctx, Int32GetDatum(fctx->lower++));
        else
            SRF_RETURN_NEXT(funcctx, Int32GetDatum(fctx->upper--));
    } else
        /* done when there are no more elements left */
        SRF_RETURN_DONE(funcctx);
}

/*
 * generate_subscripts_nodir
 *		Implements the 2-argument version of generate_subscripts
 */
Datum generate_subscripts_nodir(PG_FUNCTION_ARGS)
{
    /* just call the other one -- it can handle both cases */
    return generate_subscripts(fcinfo);
}

/*
 * array_fill_with_lower_bounds
 *		Create and fill array with defined lower bounds.
 */
Datum array_fill_with_lower_bounds(PG_FUNCTION_ARGS)
{
    ArrayType* dims = NULL;
    ArrayType* lbs = NULL;
    ArrayType* result = NULL;
    Oid elmtype;
    Datum value;
    bool isnull = false;

    if (PG_ARGISNULL(1) || PG_ARGISNULL(2))
        ereport(ERROR,
            (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED), errmsg("dimension array or low bound array cannot be null")));

    dims = PG_GETARG_ARRAYTYPE_P(1);
    lbs = PG_GETARG_ARRAYTYPE_P(2);

    if (!PG_ARGISNULL(0)) {
        value = PG_GETARG_DATUM(0);
        isnull = false;
    } else {
        value = 0;
        isnull = true;
    }

    elmtype = get_fn_expr_argtype(fcinfo->flinfo, 0);
    if (!OidIsValid(elmtype))
        ereport(ERROR, (errcode(ERRCODE_DATATYPE_MISMATCH), errmsg("could not determine data type of input")));

    result = array_fill_internal(dims, lbs, value, isnull, elmtype, fcinfo);
    PG_RETURN_ARRAYTYPE_P(result);
}

/*
 * array_fill
 *		Create and fill array with default lower bounds.
 */
Datum array_fill(PG_FUNCTION_ARGS)
{
    ArrayType* dims = NULL;
    ArrayType* result = NULL;
    Oid elmtype;
    Datum value;
    bool isnull = false;

    if (PG_ARGISNULL(1))
        ereport(ERROR,
            (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED), errmsg("dimension array or low bound array cannot be null")));

    dims = PG_GETARG_ARRAYTYPE_P(1);

    if (!PG_ARGISNULL(0)) {
        value = PG_GETARG_DATUM(0);
        isnull = false;
    } else {
        value = 0;
        isnull = true;
    }

    elmtype = get_fn_expr_argtype(fcinfo->flinfo, 0);
    if (!OidIsValid(elmtype))
        ereport(ERROR, (errcode(ERRCODE_DATATYPE_MISMATCH), errmsg("could not determine data type of input")));

    result = array_fill_internal(dims, NULL, value, isnull, elmtype, fcinfo);
    PG_RETURN_ARRAYTYPE_P(result);
}

static ArrayType* create_array_envelope(int ndims, int* dimv, const int* lbsv, int nbytes, Oid elmtype, int dataoffset)
{
    ArrayType* result = NULL;

    result = (ArrayType*)palloc0(nbytes);
    SET_VARSIZE(result, nbytes);
    result->ndim = ndims;
    result->dataoffset = dataoffset;
    result->elemtype = elmtype;
    errno_t errorno = EOK;
    errorno = memcpy_s(ARR_DIMS(result), ndims * sizeof(int), dimv, ndims * sizeof(int));
    securec_check(errorno, "\0", "\0");
    errorno = memcpy_s(ARR_LBOUND(result), ndims * sizeof(int), lbsv, ndims * sizeof(int));
    securec_check(errorno, "\0", "\0");

    return result;
}

static ArrayType* array_fill_internal(
    ArrayType* dims, ArrayType* lbs, Datum value, bool isnull, Oid elmtype, FunctionCallInfo fcinfo)
{
    ArrayType* result = NULL;
    int* dimv = NULL;
    int* lbsv = NULL;
    int ndims;
    int nitems;
    int deflbs[MAXDIM];
    int16 elmlen;
    bool elmbyval = false;
    char elmalign;
    ArrayMetaState* my_extra = NULL;

    /*
     * Params checks
     */
    if (ARR_NDIM(dims) != 1)
        ereport(ERROR,
            (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
                errmsg("wrong number of array subscripts"),
                errdetail("Dimension array must be one dimensional.")));

    if (ARR_LBOUND(dims)[0] != 1)
        ereport(ERROR,
            (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
                errmsg("wrong range of array subscripts"),
                errdetail("Lower bound of dimension array must be one.")));

    if (array_contains_nulls(dims))
        ereport(ERROR, (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED), errmsg("dimension values cannot be null")));

    dimv = (int*)ARR_DATA_PTR(dims);
    ndims = ARR_DIMS(dims)[0];

    if (ndims < 0) /* we do allow zero-dimension arrays */
        ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("invalid number of dimensions: %d", ndims)));
    if (ndims > MAXDIM)
        ereport(ERROR,
            (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                errmsg("number of array dimensions (%d) exceeds the maximum allowed (%d)", ndims, MAXDIM)));

    if (lbs != NULL) {
        if (ARR_NDIM(lbs) != 1)
            ereport(ERROR,
                (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
                    errmsg("wrong number of array subscripts"),
                    errdetail("Dimension array must be one dimensional.")));

        if (ARR_LBOUND(lbs)[0] != 1)
            ereport(ERROR,
                (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
                    errmsg("wrong range of array subscripts"),
                    errdetail("Lower bound of dimension array must be one.")));

        if (array_contains_nulls(lbs))
            ereport(ERROR, (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED), errmsg("dimension values cannot be null")));

        if (ARR_DIMS(lbs)[0] != ndims)
            ereport(ERROR,
                (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
                    errmsg("wrong number of array subscripts"),
                    errdetail("Low bound array has different size than dimensions array.")));

        lbsv = (int*)ARR_DATA_PTR(lbs);
    } else {
        int i;

        for (i = 0; i < MAXDIM; i++)
            deflbs[i] = 1;

        lbsv = deflbs;
    }

    /* fast track for empty array */
    if (ndims == 0)
        return construct_empty_array(elmtype);

    nitems = ArrayGetNItems(ndims, dimv);
    ArrayCheckBounds(ndims, dimv, lbsv);

    /*
     * We arrange to look up info about element type only once per series of
     * calls, assuming the element type doesn't change underneath us.
     */
    my_extra = (ArrayMetaState*)fcinfo->flinfo->fn_extra;
    if (my_extra == NULL) {
        fcinfo->flinfo->fn_extra = MemoryContextAlloc(fcinfo->flinfo->fn_mcxt, sizeof(ArrayMetaState));
        my_extra = (ArrayMetaState*)fcinfo->flinfo->fn_extra;
        my_extra->element_type = InvalidOid;
    }

    if (my_extra->element_type != elmtype) {
        /* Get info about element type */
        get_typlenbyvalalign(elmtype, &my_extra->typlen, &my_extra->typbyval, &my_extra->typalign);
        my_extra->element_type = elmtype;
    }

    elmlen = my_extra->typlen;
    elmbyval = my_extra->typbyval;
    elmalign = my_extra->typalign;

    /* compute required space */
    if (!isnull) {
        int i;
        char* p = NULL;
        int nbytes;
        int totbytes;

        /* make sure data is not toasted */
        if (elmlen == -1)
            value = PointerGetDatum(PG_DETOAST_DATUM(value));

        nbytes = att_addlength_datum(0, elmlen, value);
        nbytes = att_align_nominal(nbytes, elmalign);
        Assert(nbytes > 0);

        totbytes = nbytes * nitems;

        /* check for overflow of multiplication or total request */
        if (totbytes / nbytes != nitems || !AllocSizeIsValid(totbytes))
            ereport(ERROR,
                (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                    errmsg("array size exceeds the maximum allowed (%d)", (int)MaxAllocSize)));

        /*
         * This addition can't overflow, but it might cause us to go past
         * MaxAllocSize.  We leave it to palloc to complain in that case.
         */
        totbytes += ARR_OVERHEAD_NONULLS(ndims);

        result = create_array_envelope(ndims, dimv, lbsv, totbytes, elmtype, 0);

        p = ARR_DATA_PTR(result);
        for (i = 0; i < nitems; i++)
            p += ArrayCastAndSet(value, elmlen, elmbyval, elmalign, p);
    } else {
        int nbytes;
        int dataoffset;

        dataoffset = ARR_OVERHEAD_WITHNULLS(ndims, nitems);
        nbytes = dataoffset;

        result = create_array_envelope(ndims, dimv, lbsv, nbytes, elmtype, dataoffset);

        /* create_array_envelope already zeroed the bitmap, so we're done */
    }

    return result;
}

/*
 * UNNEST
 */
Datum array_unnest(PG_FUNCTION_ARGS)
{
    typedef struct {
        ArrayType* arr;
        int nextelem;
        int numelems;
        char* elemdataptr;    /* this moves with nextelem */
        bits8* arraynullsptr; /* this does not */
        int16 elmlen;
        bool elmbyval;
        char elmalign;
    } array_unnest_fctx;

    FuncCallContext* funcctx = NULL;
    array_unnest_fctx* fctx = NULL;
    MemoryContext oldcontext;

    /* stuff done only on the first call of the function */
    if (SRF_IS_FIRSTCALL()) {
        ArrayType* arr = NULL;

        /* create a function context for cross-call persistence */
        funcctx = SRF_FIRSTCALL_INIT();

        /*
         * switch to memory context appropriate for multiple function calls
         */
        oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);

        /*
         * Get the array value and detoast if needed.  We can't do this
         * earlier because if we have to detoast, we want the detoasted copy
         * to be in multi_call_memory_ctx, so it will go away when we're done
         * and not before.	(If no detoast happens, we assume the originally
         * passed array will stick around till then.)
         */
        arr = PG_GETARG_ARRAYTYPE_P(0);

        /* allocate memory for user context */
        fctx = (array_unnest_fctx*)palloc(sizeof(array_unnest_fctx));

        /* initialize state */
        fctx->arr = arr;
        fctx->nextelem = 0;
        fctx->numelems = ArrayGetNItems(ARR_NDIM(arr), ARR_DIMS(arr));

        fctx->elemdataptr = ARR_DATA_PTR(arr);
        fctx->arraynullsptr = ARR_NULLBITMAP(arr);

        get_typlenbyvalalign(ARR_ELEMTYPE(arr), &fctx->elmlen, &fctx->elmbyval, &fctx->elmalign);

        funcctx->user_fctx = fctx;
        MemoryContextSwitchTo(oldcontext);
    }

    /* stuff done on every call of the function */
    funcctx = SRF_PERCALL_SETUP();
    fctx = (array_unnest_fctx*)funcctx->user_fctx;

    if (fctx->nextelem < fctx->numelems) {
        int offset = fctx->nextelem++;
        Datum elem;

        /*
         * Check for NULL array element
         */
        if (array_get_isnull(fctx->arraynullsptr, offset)) {
            fcinfo->isnull = true;
            elem = (Datum)0;
            /* elemdataptr does not move */
        } else {
            /*
             * OK, get the element
             */
            char* ptr = fctx->elemdataptr;

            fcinfo->isnull = false;
            elem = ArrayCast(ptr, fctx->elmbyval, fctx->elmlen);

            /*
             * Advance elemdataptr over it
             */
            ptr = att_addlength_pointer(ptr, fctx->elmlen, ptr);
            ptr = (char*)att_align_nominal(ptr, fctx->elmalign);
            fctx->elemdataptr = ptr;
        }

        SRF_RETURN_NEXT(funcctx, elem);
    } else {
        /* do when there is no more left */
        SRF_RETURN_DONE(funcctx);
    }
}

/*
 * The type, dimension needs to be checked during collection operations.
 */
static void array_multiset_check(const ArrayType* v1, const ArrayType* v2)
{
    int ndims1 = (v1 == NULL) ? 0 : ARR_NDIM(v1);
    int ndims2 = (v2 == NULL) ? 0 : ARR_NDIM(v2);
    if (v1 && v2) {
        /* check element type */
        Oid element_type1 = ARR_ELEMTYPE(v1);
        Oid element_type2 = ARR_ELEMTYPE(v2);    
        if (element_type1 != element_type2) {
            ereport(ERROR, 
                (errcode(ERRCODE_DATATYPE_MISMATCH),
                    errmsg("multiset function cannot support different element types")));
        }
    }

    /* check dimensions */
    if (ndims1 > 1 || ndims2 > 1) {
        ereport(ERROR,
            (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
                errmsg("Arrays larger than one-dimension are now not supported")));
    }
}

/* Find the number of elements in the array. */
static int numDatumInArray(FunctionCallInfoData locfcinfo, Datum elt1, bool isnull1, ArrayType* array,
    const TypeCacheEntry* typentry, bool isEarlyReturn)
{
    bool oprresult = false;
    int typlen = typentry->typlen;
    bool typbyval = typentry->typbyval;
    char typalign = typentry->typalign;
    Oid element_type = ARR_ELEMTYPE(array);
    Datum* values2 = NULL;
    bool* nulls2 = NULL;
    int nelems2 = 0;
    deconstruct_array(array, element_type, typlen, typbyval, typalign, &values2, &nulls2, &nelems2);
    int j;
    int numresult = 0;
    for (j = 0; j < nelems2; j++) {
        Datum elt2 = values2[j];
        bool isnull2 = nulls2[j];
        if (isnull1 && isnull2) {
            numresult += 1;
            if (isEarlyReturn) {
                break;
            }
        }
        if (isnull1 || isnull2) {
            continue;
        }

        /*
        * Apply the operator to the element pair
        */
        locfcinfo.arg[0] = elt1;
        locfcinfo.arg[1] = elt2;
        locfcinfo.argnull[0] = false;
        locfcinfo.argnull[1] = false;
        locfcinfo.isnull = false;
        oprresult = DatumGetBool(FunctionCallInvoke(&locfcinfo));
        if (oprresult) {
            numresult += 1;
            if (isEarlyReturn) {
                break;
            }
        }
    }

    pfree(values2);
    pfree(nulls2);
    return numresult;
}

/*
 * This function is only called when the element first appears.
 * The first occurrence of the element is startIndex. So we're going to search from the 
 * next location (startIndex + 1) to see if there's the same element
 */
static int numDatumInArratByIndex(FunctionCallInfoData locfcinfo, Datum elt1, bool isnull1,
    int startIndex, Datum* values2, const bool* nulls2, int nelems2)
{
    /* elt1 is values2[startIndex] */
    int num = 1;
    int j;
    for (j = startIndex + 1; j < nelems2; j++) {
        Datum elt2 = values2[j];
        bool isnull2 = nulls2[j];
        if (isnull1 && isnull2) {
            num += 1;
        }
        if (isnull1 || isnull2) {
            continue;
        }
        /*
        * Apply the operator to the element pair
        */
        locfcinfo.arg[0] = elt1;
        locfcinfo.arg[1] = elt2;
        locfcinfo.argnull[0] = false;
        locfcinfo.argnull[1] = false;
        locfcinfo.isnull = false;
        if (DatumGetBool(FunctionCallInvoke(&locfcinfo))) {
            num += 1;
        }
    }
    return num;
}

/*
 * Insert array A into array B and deduplicate array B.
 */
static ArrayType* arrayInsertDistinctArray(ArrayType* result, ArrayType* src, TypeCacheEntry* typentry, int* index)
{
    int typlen = typentry->typlen;
    bool typbyval = typentry->typbyval;
    char typalign = typentry->typalign;
    /*
     * Apply the comparison operator to each pair of array elements.
     */
    FunctionCallInfoData locfcinfo;
    InitFunctionCallInfoData(locfcinfo, &typentry->eq_opr_finfo, MULTISET_ARGS_NUM, 0, NULL, NULL);
    /* Loop over source data */
    int nelems1 = ArrayGetNItems(ARR_NDIM(src), ARR_DIMS(src));
    char* ptr1 = ARR_DATA_PTR(src);
    bits8* bitmap1 = ARR_NULLBITMAP(src);
    uint32 bitmask = 1;
    int i;

    for (i = 0; i < nelems1; i++) {
        Datum elt1;
        bool isnull1 = false;

        /* Get element, checking for NULL */
        if (bitmap1 && (*bitmap1 & bitmask) == 0) {
            isnull1 = true;
            elt1 = (Datum)0;
        } else {
            isnull1 = false;
            elt1 = fetch_att(ptr1, typbyval, typlen);
            ptr1 = att_addlength_pointer(ptr1, typlen, ptr1);
            ptr1 = (char*)att_align_nominal(ptr1, typalign);
        }

        /* advance bitmap pointer if any */
        if (bitmap1 != NULL) {
            bitmask <<= 1;
            if (bitmask == 0x100) {
                bitmap1++;
                bitmask = 1;
            }
        }

        if (numDatumInArray(locfcinfo, elt1, isnull1, result, typentry, true) <= 0) {
            result = array_set(result, 1, index, elt1, isnull1,  -1, typlen, typbyval, typalign);
            *index += 1;
        }
    }
    
    return result;
}

/* 
 * ARRAY1 = [1,2,3]
 * ARRAY2 = [3,4,5]
 * [1,2,3,3,4,5] = array_union(ARRAY1, ARRAY2)
 * Note: only support 1 dimensional arrays
 */
Datum array_union(PG_FUNCTION_ARGS)
{
    ArrayType *v1 = NULL, *v2 = NULL;
    ArrayType* result = NULL;
    int *dims = NULL, *lbs = NULL, ndims, nitems, ndatabytes, nbytes;
    int *dims1 = NULL, *lbs1 = NULL, ndims1, nitems1, ndatabytes1;
    int *dims2 = NULL, *lbs2 = NULL, ndims2, nitems2, ndatabytes2;
    char *dat1 = NULL, *dat2 = NULL;
    bits8 *bitmap1 = NULL, *bitmap2 = NULL;
    Oid element_type;
    int32 dataoffset;
    errno_t rc = EOK;

    /* Concatenating a null array is a no-op, just return the other input */
    if (PG_ARGISNULL(0)) {
        if (PG_ARGISNULL(1))
            PG_RETURN_NULL();
        result = PG_GETARG_ARRAYTYPE_P(1);
        PG_RETURN_ARRAYTYPE_P(result);
    }
    if (PG_ARGISNULL(1)) {
        result = PG_GETARG_ARRAYTYPE_P(0);
        PG_RETURN_ARRAYTYPE_P(result);
    }

    v1 = PG_GETARG_ARRAYTYPE_P(0);
    v2 = PG_GETARG_ARRAYTYPE_P(1);

    array_multiset_check(v1, v2);

    /* OK, use it */
    element_type = ARR_ELEMTYPE(v1);

    ndims1 = ARR_NDIM(v1);
    ndims2 = ARR_NDIM(v2);

    /*
     * short circuit - if one input array is empty, and the other is not, we
     * return the non-empty one as the result
     *
     * if both are empty, return the first one
     */
    if (ndims1 == 0 && ndims2 > 0)
        PG_RETURN_ARRAYTYPE_P(v2);

    if (ndims2 == 0)
        PG_RETURN_ARRAYTYPE_P(v1);

    if (ndims1 != ndims2)
        ereport(ERROR,
            (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
                errmsg("cannot union incompatible arrays"),
                errdetail("Arrays of %d and %d dimensions are not "
                          "compatible for concatenation.",
                    ndims1,
                    ndims2)));

    /* get argument array details */
    lbs1 = ARR_LBOUND(v1);
    lbs2 = ARR_LBOUND(v2);
    dims1 = ARR_DIMS(v1);
    dims2 = ARR_DIMS(v2);
    dat1 = ARR_DATA_PTR(v1);
    dat2 = ARR_DATA_PTR(v2);
    bitmap1 = ARR_NULLBITMAP(v1);
    bitmap2 = ARR_NULLBITMAP(v2);
    nitems1 = ArrayGetNItems(ndims1, dims1);
    nitems2 = ArrayGetNItems(ndims2, dims2);
    ndatabytes1 = ARR_SIZE(v1) - ARR_DATA_OFFSET(v1);
    ndatabytes2 = ARR_SIZE(v2) - ARR_DATA_OFFSET(v2);

    /*
     * resulting array is made up of the elements (possibly arrays
     * themselves) of the input argument arrays
     */
    ndims = ndims1;
    dims = (int*)palloc(ndims * sizeof(int));
    lbs = (int*)palloc(ndims * sizeof(int));
    /* dims1[0] is elements nums of dimension 1 in the left array */
    if (unlikely(INT_MAX - dims1[0] < dims2[0])) {
        ereport(ERROR,
            (errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
                errmsg("cannot accpect arrays with dimensions out of range")));
    }
    dims[0] = dims1[0] + dims2[0];
    /* start index must be 1 in multiset union */
    lbs[0] = 1;

    /* Do this mainly for overflow checking */
    nitems = ArrayGetNItems(ndims, dims);

    /* build the result array */
    ndatabytes = ndatabytes1 + ndatabytes2;
    if (ARR_HASNULL(v1) || ARR_HASNULL(v2)) {
        dataoffset = ARR_OVERHEAD_WITHNULLS(ndims, nitems);
        nbytes = ndatabytes + dataoffset;
    } else {
        dataoffset = 0; /* marker for no null bitmap */
        nbytes = ndatabytes + ARR_OVERHEAD_NONULLS(ndims);
    }
    result = (ArrayType*)palloc0(nbytes);
    SET_VARSIZE(result, nbytes);
    result->ndim = ndims;
    result->dataoffset = dataoffset;
    result->elemtype = element_type;
    rc = memcpy_s(ARR_DIMS(result), ndims * sizeof(int), dims, ndims * sizeof(int));
    securec_check(rc, "", "");
    rc = memcpy_s(ARR_LBOUND(result), ndims * sizeof(int), lbs, ndims * sizeof(int));
    securec_check(rc, "", "");

    pfree_ext(dims);
    pfree_ext(lbs);

    /* data area is arg1 then arg2. And make sure the destMax of memcpy_s should never be zero. */
    if (ndatabytes1 > 0) {
        rc = memcpy_s(ARR_DATA_PTR(result), ndatabytes1, dat1, ndatabytes1);
        securec_check(rc, "", "");
    }
    if (ndatabytes2 > 0) {
        rc = memcpy_s(ARR_DATA_PTR(result) + ndatabytes1, ndatabytes2, dat2, ndatabytes2);
        securec_check(rc, "", "");
    }

    /* handle the null bitmap if needed */
    if (ARR_HASNULL(result)) {
        array_bitmap_copy(ARR_NULLBITMAP(result), 0, bitmap1, 0, nitems1);
        array_bitmap_copy(ARR_NULLBITMAP(result), nitems1, bitmap2, 0, nitems2);
    }

    PG_RETURN_ARRAYTYPE_P(result);
}

/* 
 * ARRAY1 = [1,2,3]
 * ARRAY2 = [3,4,5]
 * [1,2,3,4,5] = array_union_distinct(ARRAY1, ARRAY2)
 * Note: only support 1 dimensional arrays
 */
Datum array_union_distinct(PG_FUNCTION_ARGS)
{
    ArrayType* v1 = NULL;
    ArrayType* v2 = NULL;
    Oid element_type = InvalidOid;

    if (PG_ARGISNULL(0) && PG_ARGISNULL(1)) {
        PG_RETURN_NULL();
    }
    if (PG_ARGISNULL(0)) {
        v1 = NULL;
    } else {
        v1 = PG_GETARG_ARRAYTYPE_P(0);
        element_type = ARR_ELEMTYPE(v1);
    }
    if (PG_ARGISNULL(1)) {
        v2 = NULL;
    } else {
        v2 = PG_GETARG_ARRAYTYPE_P(1);
        element_type = ARR_ELEMTYPE(v2);
    }
    
    array_multiset_check(v1, v2);
    TypeCacheEntry* typentry = (TypeCacheEntry*)fcinfo->flinfo->fn_extra;
    if (typentry == NULL || typentry->type_id != element_type) {
        typentry = lookup_type_cache(element_type, TYPECACHE_EQ_OPR_FINFO);
        if (!OidIsValid(typentry->eq_opr_finfo.fn_oid))
            ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_FUNCTION),
                    errmsg("could not identify an equality operator for type %s", format_type_be(element_type))));
    }
    
    ArrayType* result = construct_empty_array(element_type);
    int index = 1;
    if (v1 != NULL && ARR_NDIM(v1) == 1) {
        result = arrayInsertDistinctArray(result, v1, typentry, &index);
    }

    if (v2 != NULL && ARR_NDIM(v2) == 1) {
        result = arrayInsertDistinctArray(result, v2, typentry, &index);
    }

    PG_RETURN_ARRAYTYPE_P(result);
}

static ArrayType* array_intersect_internal(ArrayType* v1, ArrayType* v2, TypeCacheEntry* typentry, bool isDistinct)
{
    Oid element_type = ARR_ELEMTYPE(v1);
    if (typentry == NULL || typentry->type_id != element_type) {
        typentry = lookup_type_cache(element_type, TYPECACHE_EQ_OPR_FINFO);
        if (!OidIsValid(typentry->eq_opr_finfo.fn_oid))
            ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_FUNCTION),
                    errmsg("could not identify an equality operator for type %s", format_type_be(element_type))));
    }
    /*
     * Apply the comparison operator to each pair of array elements.
     */
    FunctionCallInfoData locfcinfo;
    InitFunctionCallInfoData(locfcinfo, &typentry->eq_opr_finfo, MULTISET_ARGS_NUM, 0, NULL, NULL);
    int typlen = typentry->typlen;
    bool typbyval = typentry->typbyval;
    char typalign = typentry->typalign;
    Datum* values1 = NULL;
    bool* nulls1 = NULL;
    int nelems1 = 0;
    deconstruct_array(v1, element_type, typlen, typbyval, typalign, &values1, &nulls1, &nelems1);
    ArrayType* result = construct_empty_array(element_type);
    int i;
    int index = 1;
    for (i = 0; i < nelems1; i++) {
        Datum elt1 = values1[i];
        bool isnull1 = nulls1[i];
        /* if this element is already in result, continue */
        if (numDatumInArray(locfcinfo, elt1, isnull1, result, typentry, true) > 0) {
            continue;
        }

        if (isDistinct) {
            /* find the first same element, and return */
            if (numDatumInArray(locfcinfo, elt1, isnull1, v2, typentry, true) > 0) {
                result = array_set(result, 1, &index, elt1, isnull1,  -1, typlen, typbyval, typalign);
                index += 1;
            }
        } else {
            /* find the num of the same element in array2 */
            int numInArray2 = numDatumInArray(locfcinfo, elt1, isnull1, v2, typentry, false);
            if (numInArray2) {
                /* find the num of the same element in array1 */
                int numInArray1 = 
                    numDatumInArratByIndex(locfcinfo, elt1, isnull1, i, values1, nulls1, nelems1);
                int minNum = (numInArray1 > numInArray2) ? numInArray2 : numInArray1;
                for (int k = 0; k < minNum; k++) {
                    result = array_set(result, 1, &index, elt1, isnull1,  -1, typlen, typbyval, typalign);
                    index += 1;
                }
            }
        }
    }
    return result;
}

/*
 * ARRAY1 = [1,2,3,3,NULL,NULL]
 * ARRAY2 = [3,3,4,5,NULL,NULL]
 * [3,3,NULL,NULL] = array_intersect(ARRAY1, ARRAY2)
 */
Datum array_intersect(PG_FUNCTION_ARGS)
{
    if (PG_ARGISNULL(0) || PG_ARGISNULL(1)) {
        PG_RETURN_NULL();
    }
    ArrayType* v1 = PG_GETARG_ARRAYTYPE_P(0);
    ArrayType* v2 = PG_GETARG_ARRAYTYPE_P(1);
    array_multiset_check(v1, v2);
    TypeCacheEntry* typentry = (TypeCacheEntry*)fcinfo->flinfo->fn_extra;
    ArrayType* result = array_intersect_internal(v1, v2, typentry, false);
    PG_RETURN_ARRAYTYPE_P(result);
}

/*
 * ARRAY1 = [1,2,3,3,NULL,NULL]
 * ARRAY2 = [3,3,4,5,NULL,NULL]
 * [3,NULL] = array_intersect_distinct(ARRAY1, ARRAY2)
 */
Datum array_intersect_distinct(PG_FUNCTION_ARGS)
{
    if (PG_ARGISNULL(0) || PG_ARGISNULL(1)) {
        PG_RETURN_NULL();
    }
    ArrayType* v1 = PG_GETARG_ARRAYTYPE_P(0);
    ArrayType* v2 = PG_GETARG_ARRAYTYPE_P(1);
    array_multiset_check(v1, v2);
    TypeCacheEntry* typentry = (TypeCacheEntry*)fcinfo->flinfo->fn_extra;
    ArrayType* result = array_intersect_internal(v1, v2, typentry, true);
    PG_RETURN_ARRAYTYPE_P(result);
}

static ArrayType* array_except_internal(ArrayType* v1, ArrayType* v2, TypeCacheEntry* typentry, bool isDistinct)
{
    Oid element_type = ARR_ELEMTYPE(v1);
    if (typentry == NULL || typentry->type_id != element_type) {
        typentry = lookup_type_cache(element_type, TYPECACHE_EQ_OPR_FINFO);
        if (!OidIsValid(typentry->eq_opr_finfo.fn_oid))
            ereport(ERROR,
                (errcode(ERRCODE_UNDEFINED_FUNCTION),
                    errmsg("could not identify an equality operator for type %s", format_type_be(element_type))));
    }

    ArrayType* result = construct_empty_array(element_type);
    int index = 1;
    if (v2 == NULL) {
        /* must be distincted */
        Assert(isDistinct);
        result = arrayInsertDistinctArray(result, v1, typentry, &index);
        return result;
    }

    /*
     * Apply the comparison operator to each pair of array elements.
     */
    FunctionCallInfoData locfcinfo;
    InitFunctionCallInfoData(locfcinfo, &typentry->eq_opr_finfo, MULTISET_ARGS_NUM, 0, NULL, NULL);
    int typlen = typentry->typlen;
    bool typbyval = typentry->typbyval;
    char typalign = typentry->typalign;
    Datum* values1 = NULL;
    bool* nulls1 = NULL;
    int nelems1 = 0;
    deconstruct_array(v1, element_type, typlen, typbyval, typalign, &values1, &nulls1, &nelems1);
    int i;
    for (i = 0; i < nelems1; i++) {
        Datum elt1 = values1[i];
        bool isnull1 = nulls1[i];
        /* if this element is already in result, continue */
        if (numDatumInArray(locfcinfo, elt1, isnull1, result, typentry, true) > 0) {
            continue;
        }
        /* find the num of the same element in array2 */
        int numInArray2 = numDatumInArray(locfcinfo, elt1, isnull1, v2, typentry, false);
        int insertNum = 0;
        if (numInArray2) {
            if (isDistinct) {
                continue;
            } else {
                int numInArray1 = 
                    numDatumInArratByIndex(locfcinfo, elt1, isnull1, i, values1, nulls1, nelems1);
                insertNum = numInArray1 - numInArray2;
            }
        } else {
            if (isDistinct) {
                /* distinct only insert 1 element */
                insertNum = 1;
            } else {
                /* non-distinct should insert all same element */
                insertNum = numDatumInArratByIndex(locfcinfo, elt1, isnull1, i, values1, nulls1, nelems1);
            }
        }

        for (int k = 0; k < insertNum; k++) {
            result = array_set(result, 1, &index, elt1, isnull1,  -1, typlen, typbyval, typalign);
            index += 1;
        }
    }
    return result;
}

/*
 * ARRAY1 = [1,2,3,3,NULL,NULL]
 * ARRAY2 = [3,4,5,NULL]
 * [1,2,3,NULL] = array_except(ARRAY1, ARRAY2)
 */
Datum array_except(PG_FUNCTION_ARGS)
{
    if (PG_ARGISNULL(0)) {
        PG_RETURN_NULL();
    }
    ArrayType* v1 = PG_GETARG_ARRAYTYPE_P(0);
    if (PG_ARGISNULL(1)) {
        PG_RETURN_ARRAYTYPE_P(v1);
    }
    ArrayType* v2 = PG_GETARG_ARRAYTYPE_P(1);
    array_multiset_check(v1, v2);
    TypeCacheEntry* typentry = (TypeCacheEntry*)fcinfo->flinfo->fn_extra;
    ArrayType* result = array_except_internal(v1, v2, typentry, false);
    PG_RETURN_ARRAYTYPE_P(result);
}

/*
 * ARRAY1 = [1,2,3,3,NULL,NULL]
 * ARRAY2 = [3,4,5,NULL]
 * [1,2] = array_except(ARRAY1, ARRAY2)
 */
Datum array_except_distinct(PG_FUNCTION_ARGS)
{
    if (PG_ARGISNULL(0)) {
        PG_RETURN_NULL();
    }
    ArrayType* v1 = PG_GETARG_ARRAYTYPE_P(0);
    ArrayType* v2 = NULL;
    if (!PG_ARGISNULL(1)) {
        v2 = PG_GETARG_ARRAYTYPE_P(1);
    }
    array_multiset_check(v1, v2);
    TypeCacheEntry* typentry = (TypeCacheEntry*)fcinfo->flinfo->fn_extra;
    ArrayType* result = array_except_internal(v1, v2, typentry, true);
    PG_RETURN_ARRAYTYPE_P(result);
}

/*
 * check if search is same as replace
 */
static bool array_same_replace(FunctionCallInfo locfcinfo, Datum search, bool search_isnull,
                               Datum replace, bool replace_isnull)
{
    if (search_isnull != replace_isnull) {
        return false;
    } else if (search_isnull) {
        Assert(replace_isnull);
        return true;
    } else {
        locfcinfo->arg[0] = search;
        locfcinfo->arg[1] = replace;
        locfcinfo->argnull[0] = search_isnull;
        locfcinfo->argnull[1] = replace_isnull;
        locfcinfo->isnull = false;
        return DatumGetBool(FunctionCallInvoke(locfcinfo));
    }
}

/*
 * array_replace/array_remove support
 *
 * Find all array entries matching (not distinct from) search/search_isnull,
 * and delete them if remove is true, else replace them with
 * replace/replace_isnull.  Comparisons are done using the specified
 * collation.  fcinfo is passed only for caching purposes.
 */
static ArrayType *array_replace_internal(ArrayType *array, Datum search, bool search_isnull, Datum replace,
                                         bool replace_isnull, bool remove, Oid collation, FunctionCallInfo fcinfo)
{
    ArrayType *result = NULL;
    Oid        element_type;
    Datum     *values = NULL;
    bool      *nulls = NULL;
    int       *dim = NULL;
    int        ndim;
    int        nitems;
    int        nresult;
    int        i;
    int32      nbytes = 0;
    int32      dataoffset;
    bool       hasnulls = false;
    int        typlen;
    bool       typbyval = false;
    char       typalign;
    char      *arraydataptr = NULL;
    bits8     *bitmap = NULL;
    int        bitmask;
    bool       changed = false;
    TypeCacheEntry *typentry = NULL;
    FunctionCallInfoData locfcinfo;
    errno_t rc = EOK;
    
    element_type = ARR_ELEMTYPE(array);
    ndim = ARR_NDIM(array);
    dim = ARR_DIMS(array);
    nitems = ArrayGetNItems(ndim, dim);

    /* Return input array unmodified if it is empty */
    if (nitems <= 0)
        return array;

    /*
     * We can't remove elements from multi-dimensional arrays, since the
     * result might not be rectangular.
     */
    if (remove && ndim > 1)
        ereport(ERROR,
                (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                 errmsg("removing elements from multidimensional arrays is not supported")));

    /*
     * We arrange to look up the equality function only once per series of
     * calls, assuming the element type doesn't change underneath us.
     */
    typentry = (TypeCacheEntry *) fcinfo->flinfo->fn_extra;
    if (typentry == NULL || typentry->type_id != element_type) {
        typentry = lookup_type_cache(element_type, TYPECACHE_EQ_OPR_FINFO);
        if (!OidIsValid(typentry->eq_opr_finfo.fn_oid))
            ereport(ERROR,
                    (errcode(ERRCODE_UNDEFINED_FUNCTION),
                     errmsg("could not identify an equality operator for type %s",
                            format_type_be(element_type))));
        fcinfo->flinfo->fn_extra = (void *) typentry;
    }
    typlen = typentry->typlen;
    typbyval = typentry->typbyval;
    typalign = typentry->typalign;

    /*
     * Detoast values if they are toasted.  The replacement value must be
     * detoasted for insertion into the result array, while detoasting the
     * search value only once saves cycles.
     */
    if (typlen == -1) {
        if (!search_isnull)
            search = PointerGetDatum(PG_DETOAST_DATUM(search));
        if (!replace_isnull)
            replace = PointerGetDatum(PG_DETOAST_DATUM(replace));
    }

    /* Prepare to apply the comparison operator */
    InitFunctionCallInfoData(locfcinfo, &typentry->eq_opr_finfo, 2, collation, NULL, NULL);

    /* directly return if search is same as replace */
    if (!remove && array_same_replace(&locfcinfo, search, search_isnull, replace, replace_isnull)) {
        return array;
    }

    /* Allocate temporary arrays for new values */
    values = (Datum *) palloc(nitems * sizeof(Datum));
    nulls = (bool *) palloc(nitems * sizeof(bool));

    /* Loop over source data */
    arraydataptr = ARR_DATA_PTR(array);
    bitmap = ARR_NULLBITMAP(array);
    bitmask = 1;
    hasnulls = false;
    nresult = 0;

    for (i = 0; i < nitems; i++) {
        Datum        elt;
        bool        isNull = false;
        bool        oprresult = false;
        bool        skip = false;

        /* Get source element, checking for NULL */
        if (bitmap && (*bitmap & bitmask) == 0) {
            isNull = true;
            /* If searching for NULL, we have a match */
            if (search_isnull) {
                if (remove) {
                    skip = true;
                    changed = true;
                } else if (!replace_isnull) {
                    values[nresult] = replace;
                    isNull = false;
                    changed = true;
                }
            }
        } else {
            isNull = false;
            elt = fetch_att(arraydataptr, typbyval, typlen);
            arraydataptr = att_addlength_datum(arraydataptr, typlen, elt);
            arraydataptr = (char *) att_align_nominal(arraydataptr, typalign);

            if (search_isnull) {
                /* no match possible, keep element */
                values[nresult] = elt;
            } else {
                /* Compare the pair of elements */
                locfcinfo.arg[0] = elt;
                locfcinfo.arg[1] = search;
                locfcinfo.argnull[0] = false;
                locfcinfo.argnull[1] = false;
                locfcinfo.isnull = false;
                oprresult = DatumGetBool(FunctionCallInvoke(&locfcinfo));
                if (locfcinfo.isnull || !oprresult) {
                    /* no match, keep element */
                    values[nresult] = elt;
                } else {
                    /* match, so replace or delete */
                    changed = true;
                    if (remove) {
                        skip = true;
                    } else {
                        values[nresult] = replace;
                        isNull = replace_isnull;
                    }
                }
            }
        }

        if (!skip) {
            nulls[nresult] = isNull;
            if (isNull) {
                hasnulls = true;
            } else {
                /* Update total result size */
                nbytes = att_addlength_datum(nbytes, typlen, values[nresult]);
                nbytes = att_align_nominal(nbytes, typalign);
                /* check for overflow of total request */
                if (!AllocSizeIsValid(nbytes))
                    ereport(ERROR,
                            (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
                             errmsg("array size exceeds the maximum allowed (%d)",
                                    (int) MaxAllocSize)));
            }
            nresult++;
        }

        /* advance bitmap pointer if any */
        if (bitmap) {
            bitmask <<= 1;
            if (bitmask == 0x100) {
                bitmap++;
                bitmask = 1;
            }
        }
    }

    /*
     * If not changed just return the original array
     */
    if (!changed) {
        pfree(values);
        pfree(nulls);
        return array;
    }

    /* If all elements were removed return an empty array */
    if (nresult == 0) {
        pfree(values);
        pfree(nulls);
        return construct_empty_array(element_type);
    }

    /* Allocate and initialize the result array */
    if (hasnulls) {
        dataoffset = ARR_OVERHEAD_WITHNULLS(ndim, nresult);
        nbytes += dataoffset;
    } else {
        dataoffset = 0;            /* marker for no null bitmap */
        nbytes += ARR_OVERHEAD_NONULLS(ndim);
    }
    result = (ArrayType *) palloc0(nbytes);
    SET_VARSIZE(result, nbytes);
    result->ndim = ndim;
    result->dataoffset = dataoffset;
    result->elemtype = element_type;
    rc = memcpy_s(ARR_DIMS(result), nbytes - sizeof(ArrayType), ARR_DIMS(array), ndim * sizeof(int));
    securec_check(rc, "\0", "\0");
    rc = memcpy_s(ARR_LBOUND(result), nbytes - (sizeof(ArrayType) + sizeof(int) * ndim),
                  ARR_LBOUND(array), ndim * sizeof(int));
    securec_check(rc, "\0", "\0");
	
    if (remove) {
        /* Adjust the result length */
        ARR_DIMS(result)[0] = nresult;
    }

    /* Insert data into result array */
    CopyArrayEls(result, values, nulls, nresult, typlen, typbyval, typalign, false);

    pfree(values);
    pfree(nulls);

    return result;
}

/*
 * Remove any occurrences of an element from an array
 *
 * If used on a multi-dimensional array this will raise an error.
 */
Datum array_remove(PG_FUNCTION_ARGS)
{
    ArrayType  *array = NULL;
    Datum       search = PG_GETARG_DATUM(1);
    bool        search_isnull = PG_ARGISNULL(1);

    if (PG_ARGISNULL(0))
        PG_RETURN_NULL();
    array = PG_GETARG_ARRAYTYPE_P(0);

    array = array_replace_internal(array, search, search_isnull, (Datum) 0, true, true, PG_GET_COLLATION(), fcinfo);
    PG_RETURN_ARRAYTYPE_P(array);
}

/*
 * Replace any occurrences of an element in an array
 */
Datum array_replace(PG_FUNCTION_ARGS)
{
    ArrayType  *array = NULL;
    Datum       search = PG_GETARG_DATUM(1);
    bool        search_isnull = PG_ARGISNULL(1);
    Datum       replace = PG_GETARG_DATUM(2);
    bool        replace_isnull = PG_ARGISNULL(2);

    if (PG_ARGISNULL(0))
        PG_RETURN_NULL();
    array = PG_GETARG_ARRAYTYPE_P(0);

    array = array_replace_internal(array, search, search_isnull, replace, replace_isnull, false, PG_GET_COLLATION(),
                                   fcinfo);
    PG_RETURN_ARRAYTYPE_P(array);
}

